python ETL工具 pyetl_pyetl建表

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

安装

pip3 install pyetl
1

使用

数据库表之间数据同步

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()
1
2
3
4

数据库表到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()
1
2
3
4

数据库表同步es

from pyetl import Task, DatabaseReader, ElasticsearchWriter
reader = DatabaseReader("sqlite:///db1.sqlite3", table_name="source")
writer = ElasticsearchWriter(index_name="tartget")
Task(reader, writer).start()
1
2
3
4

原始表目标表字段名称不同

# 原始表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()
1
2
3
4
5
6
7

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

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

继承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 filter_function(self, record):
        """数据流过滤无效数据"""
        return True if record["flag"] else False

    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()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

Reader和Writer介绍

使用过程中有任何疑问，欢迎评论交流

项目地址pyetl