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
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/4569.html
標籤：大數據
上一篇：Hadoop的shell操作