pyetl是一個純python開發的ETL框架, 相比sqoop, datax 之類的ETL工具,pyetl可以對每個欄位添加udf函數,使得數據轉換過程更加靈活,相比專業ETL工具pyetl更輕量,純python代碼操作,更加符合開發人員習慣 安裝 pip3 install 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