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

安裝

pip3 install pyetl

使用示例

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

from pyetl import Task, DatabaseReader, DatabaseWriterreader = 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, HiveWriter2reader = 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, ElasticSearchWriterreader = DatabaseReader('sqlite:///db1.sqlite3', table_name='source')writer = ElasticSearchWriter(hosts=['localhost'], index_name='tartget')Task(reader, writer).start()

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

添加

# 原始表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類(lèi)靈活擴(kuò)展ETL任務(wù)

import jsonfrom pyetl import Task, DatabaseReader, DatabaseWriterclass 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)字典類(lèi)型返回 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è)字段類(lèi)型都轉(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ù)到文本文件

項(xiàng)目地址pyetl

總結(jié)

到此這篇關(guān)于python ETL工具 pyetl的文章就介紹到這了,更多相關(guān)python ETL工具 pyetl內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！