快捷導(dǎo)航

python ETL工具 pyetl

更新時(shí)間：2020年06月07日 09:57:49 作者：麥葉

pyetl是一個(gè)純python開(kāi)發(fā)的ETL框架，相比sqoop, datax 之類的ETL工具，pyetl可以對(duì)每個(gè)字段添加udf函數(shù)，使得數(shù)據(jù)轉(zhuǎn)換過(guò)程更加靈活，相比專業(yè)ETL工具pyetl更輕量，純python代碼操作，更加符合開(kāi)發(fā)人員習(xí)慣。這篇文章主要介紹了python ETL工具 pyetl,需要的朋友參考下

pyetl是一個(gè)純python開(kāi)發(fā)的ETL框架，相比sqoop, datax 之類的ETL工具，pyetl可以對(duì)每個(gè)字段添加udf函數(shù)，使得數(shù)據(jù)轉(zhuǎn)換過(guò)程更加靈活，相比專業(yè)ETL工具pyetl更輕量，純python代碼操作，更加符合開(kāi)發(fā)人員習(xí)慣

安裝

pip3 install pyetl

使用示例

數(shù)據(jù)庫(kù)表之間數(shù)據(jù)同步

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

數(shù)據(jù)庫(kù)表到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()

數(shù)據(jù)庫(kù)表同步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()

原始表目標(biāo)表字段名稱不同，需要添加字段映射

添加

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

字段的udf映射，對(duì)字段進(jìn)行規(guī)則校驗(yàn)、數(shù)據(jù)標(biāo)準(zhǔn)化、數(shù)據(jù)清洗等

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

繼承Task類靈活擴(kuò)展ETL任務(wù)

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):
    """通過(guò)函數(shù)的方式生成字段映射配置，使用更靈活"""
    # 以下示例將數(shù)據(jù)庫(kù)中的字段映射配置取出后轉(zhuǎn)字典類型返回
    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):
    """通過(guò)函數(shù)的方式生成字段的udf映射"""
    # 以下示例將每個(gè)字段類型都轉(zhuǎn)換為字符串
    return {col: str for col in self.columns}
   
  def apply_function(self, record):
    """數(shù)據(jù)流中對(duì)一整條數(shù)據(jù)的udf"""
    record["flag"] = int(record["id"]) % 2
    return record

  def before(self):
    """任務(wù)開(kāi)始前要執(zhí)行的操作, 如初始化任務(wù)表，創(chuàng)建目標(biāo)表等"""
    sql = "create table destination_table(id int, name varchar(100))"
    self.writer.db.execute(sql)
  
  def after(self):
    """任務(wù)完成后要執(zhí)行的操作，如更新任務(wù)狀態(tài)等"""
    sql = "update task set status='done' where name='new_task'"
    self.writer.db.execute(sql)

NewTask().start()

目前已實(shí)現(xiàn)Reader和Writer列表

Reader	介紹
DatabaseReader	支持所有關(guān)系型數(shù)據(jù)庫(kù)的讀取
FileReader	結(jié)構(gòu)化文本數(shù)據(jù)讀取，如csv文件
ExcelReader	Excel表文件讀取

Writer	介紹
DatabaseWriter	支持所有關(guān)系型數(shù)據(jù)庫(kù)的寫(xiě)入
ElasticSearchWriter	批量寫(xiě)入數(shù)據(jù)到es索引
HiveWriter	批量插入hive表
HiveWriter2	Load data方式導(dǎo)入hive表（推薦)
FileWriter	寫(xiě)入數(shù)據(jù)到文本文件