溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
小編這次要給大家分享的是淺談python中ETL工具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 | 寫入數據到文本文件 |
看完這篇關于淺談python中ETL工具pyetl的文章，如果覺得文章內容寫得不錯的話，可以把它分享出去給更多人看到。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
