Python轻量级ETL框架pyetl：从基础同步到自定义扩展

pyetl框架概述与安装

pyetl是一个基于纯Python构建的ETL（抽取、转换、加载）框架。与Sqoop或DataX等工具相比，其核心优势在于支持字段级的用户自定义函数（UDF），且整体架构更为轻量，更贴合Python开发者的使用习惯。

安装过程十分简单，只需执行以下命令：

pip3 install pyetl

核心组件与数据同步场景

pyetl通过Reader和Writer组件实现数据流转。主要支持的组件包括：

• DatabaseReader/DatabaseWriter：支持所有关系型数据库的读取与写入。
• HiveWriter/HiveWriter2：用于批量插入Hive表，其中HiveWriter2采用Load data方式，官方推荐优先使用。
• ElasticSearchWriter：支持将数据批量写入Elasticsearch索引。
• FileReader/FileWriter：用于处理CSV等结构化文本文件的读写。
• ExcelReader：专门用于读取Excel文件。

基础数据同步示例

最简单的同步场景是将源数据库表的数据直接写入目标数据库表。以下是将SQLite数据同步至另一个SQLite数据库的示例：

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，可使用HiveWriter2：

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()

同步至Elasticsearch的示例如下：

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()

字段映射与数据转换

当源表与目标表的字段名称不一致时，需通过columns参数配置映射关系。此外，利用functions参数可定义字段级的UDF，实现数据标准化、清洗或规则校验。

字段映射配置

假设源表source包含uuid和full_name字段，而目标表target对应id和name字段，配置方式如下：

# 原始表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字典，可以为特定字段指定处理函数。例如，将ID转为字符串，并去除姓名字段的前后空格：

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

高级扩展：自定义Task类

对于复杂业务需求，可以继承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()

在上述代码中，get_columns和get_functions允许动态生成映射配置；apply_function用于处理单条记录的转换逻辑；before和after分别用于任务执行前后的初始化与状态更新操作。

以上就是Python ETL工具pyetl的详细使用指南，更多关于数据同步与ETL框架的资料请关注本站其它相关文章！