python ETL工具 pyetl

 更新时间:2020年06月07日 09:57:49   作者:麦叶  
pyetl是一个纯python开发的ETL框架, 相比sqoop, datax 之类的ETL工具,pyetl可以对每个字段添加udf函数,使得数据转换过程更加灵活,相比专业ETL工具pyetl更轻量,纯python代码操作,更加符合开发人员习惯。这篇文章主要介绍了python ETL工具 pyetl,需要的朋友参考下

pyetl是一个纯python开发的ETL框架, 相比sqoop, datax 之类的ETL工具,pyetl可以对每个字段添加udf函数,使得数据转换过程更加灵活,相比专业ETL工具pyetl更轻量,纯python代码操作,更加符合开发人员习惯

安装

pip3 install pyetl

使用示例

数据库表之间数据同步

from pyetl import Task, DatabaseReader, DatabaseWriter
reader = DatabaseReader("sqlite:///db1.sqlite3", table_name="source")
writer = DatabaseWriter("sqlite:///db2.sqlite3", table_name="target")
Task(reader, writer).start()

数据库表到hive表同步

from pyetl import Task, DatabaseReader, HiveWriter2
reader = DatabaseReader("sqlite:///db1.sqlite3", table_name="source")
writer = HiveWriter2("hive://localhost:10000/default", table_name="target")
Task(reader, writer).start()

数据库表同步es

from pyetl import Task, DatabaseReader, ElasticSearchWriter
reader = DatabaseReader("sqlite:///db1.sqlite3", table_name="source")
writer = ElasticSearchWriter(hosts=["localhost"], index_name="tartget")
Task(reader, writer).start()

原始表目标表字段名称不同,需要添加字段映射

添加

# 原始表source包含uuid,full_name字段
reader = DatabaseReader("sqlite:///db.sqlite3", table_name="source")
# 目标表target包含id,name字段
writer = DatabaseWriter("sqlite:///db.sqlite3", table_name="target")
# columns配置目标表和原始表的字段映射关系
columns = {"id": "uuid", "name": "full_name"}
Task(reader, writer, columns=columns).start()

字段的udf映射,对字段进行规则校验、数据标准化、数据清洗等

# functions配置字段的udf映射,如下id转字符串,name去除前后空格
functions={"id": str, "name": lambda x: x.strip()}
Task(reader, writer, columns=columns, functions=functions).start()

继承Task类灵活扩展ETL任务

import json
from pyetl import Task, DatabaseReader, DatabaseWriter

class NewTask(Task):
  reader = DatabaseReader("sqlite:///db.sqlite3", table_name="source")
  writer = DatabaseWriter("sqlite:///db.sqlite3", table_name="target")
  
  def get_columns(self):
    """通过函数的方式生成字段映射配置,使用更灵活"""
    # 以下示例将数据库中的字段映射配置取出后转字典类型返回
    sql = "select columns from task where name='new_task'"
    columns = self.writer.db.read_one(sql)["columns"]
    return json.loads(columns)
   
  def get_functions(self):
    """通过函数的方式生成字段的udf映射"""
    # 以下示例将每个字段类型都转换为字符串
    return {col: str for col in self.columns}
   
  def apply_function(self, record):
    """数据流中对一整条数据的udf"""
    record["flag"] = int(record["id"]) % 2
    return record

  def before(self):
    """任务开始前要执行的操作, 如初始化任务表,创建目标表等"""
    sql = "create table destination_table(id int, name varchar(100))"
    self.writer.db.execute(sql)
  
  def after(self):
    """任务完成后要执行的操作,如更新任务状态等"""
    sql = "update task set status='done' where name='new_task'"
    self.writer.db.execute(sql)

NewTask().start()

目前已实现Reader和Writer列表

 

Reader 介绍
DatabaseReader 支持所有关系型数据库的读取
FileReader 结构化文本数据读取,如csv文件
ExcelReader Excel表文件读取

Writer 介绍
DatabaseWriter 支持所有关系型数据库的写入
ElasticSearchWriter 批量写入数据到es索引
HiveWriter 批量插入hive表
HiveWriter2 Load data方式导入hive表(推荐)
FileWriter 写入数据到文本文件

项目地址pyetl

总结

到此这篇关于python ETL工具 pyetl的文章就介绍到这了,更多相关python ETL工具 pyetl内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

相关文章

  • Python中selenium库的用法详解

    Python中selenium库的用法详解

    这篇文章主要介绍了Python中selenium库的用法详解,需要的朋友可以参考下
    2021-05-05
  • matlab中imadjust函数的作用及应用举例

    matlab中imadjust函数的作用及应用举例

    这篇文章主要介绍了matlab中imadjust函数的作用及应用举例,本文通过实例代码给大家介绍的非常详细,具有一定的参考借鉴价值,需要的朋友可以参考下
    2020-02-02
  • pandas中的dataframe汇总和计算方法

    pandas中的dataframe汇总和计算方法

    这篇文章主要介绍了pandas中的dataframe汇总和计算方法,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2023-06-06
  • python中with用法讲解

    python中with用法讲解

    在本篇文章里小编给大家整理的是关于python中with用法讲解内容,有需要的朋友们可以参考下。
    2020-02-02
  • Python PIL库图片灰化处理

    Python PIL库图片灰化处理

    这篇文章主要介绍了Python图片灰化处理PIL库的相关知识,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2020-04-04
  • Linux永久修改pip配置源的详细过程

    Linux永久修改pip配置源的详细过程

    默认情况下pip使用的是国外的镜像,在下载的时候速度非常慢,所以需要更换PIP的镜像源,下面这篇文章主要给大家介绍了关于Linux永久修改pip配置源的相关资料,需要的朋友可以参考下
    2024-02-02
  • python决策树之C4.5算法详解

    python决策树之C4.5算法详解

    这篇文章主要为大家详细介绍了python决策树之C4.5算法的相关资料,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2017-12-12
  • 解决Keras 中加入lambda层无法正常载入模型问题

    解决Keras 中加入lambda层无法正常载入模型问题

    这篇文章主要介绍了解决Keras 中加入lambda层无法正常载入模型问题,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2020-06-06
  • 用Python写一个无界面的2048小游戏

    用Python写一个无界面的2048小游戏

    这篇文章主要介绍了用Python写一个无界面的2048小游戏的相关资料,需要的朋友可以参考下
    2016-05-05
  • pandas数据处理之 标签列字符转数字的实现

    pandas数据处理之 标签列字符转数字的实现

    这篇文章主要介绍了pandas数据处理之 标签列字符转数字的实现方式,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2021-03-03

最新评论