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,HiveWriter2
reader = DatabaseReader("sqlite:///db1.sqlite3",table_name="source")
writer = HiveWriter2("hive://localhost:10000/default",writer).start()

資料庫表同步es

from pyetl import Task,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,columns=columns,functions=functions).start()

繼承Task類靈活擴充套件ETL任務

import json
from pyetl import Task,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內容請搜尋我們以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援我們！