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
总结
P70系列延期,华为新旗舰将在下月发布
3月20日消息,近期博主@数码闲聊站 透露,原定三月份发布的华为新旗舰P70系列延期发布,预计4月份上市。
而博主@定焦数码 爆料,华为的P70系列在定位上已经超过了Mate60,成为了重要的旗舰系列之一。它肩负着重返影像领域顶尖的使命。那么这次P70会带来哪些令人惊艳的创新呢?
根据目前爆料的消息来看,华为P70系列将推出三个版本,其中P70和P70 Pro采用了三角形的摄像头模组设计,而P70 Art则采用了与上一代P60 Art相似的不规则形状设计。这样的外观是否好看见仁见智,但辨识度绝对拉满。
更新日志
- 孙悦.1996-伙伴【正大国际】【WAV+CUE】
- 纪钧瀚《钢琴阅读时光 雨中书店聆听轻音乐》[FLAC/分轨][399.62MB]
- 证声音乐图书馆《走向自然 疗心爵士乐》[320K/MP3][87.4MB]
- 证声音乐图书馆《走向自然 疗心爵士乐》[FLAC/分轨][184.94MB]
- 陈慧娴.2018-Priscilla-Ism演唱会3CD(2024环球红馆40复刻系列)【环球】【WAV+CUE】
- 郑秀文.1999-我应该得到(国)【华纳】【WAV+CUE】
- 陈家慧.2011-钢琴酒吧2CD【龙吟唱片】【WAV+CUE】
- 证声音乐图书馆《雨季 蓝调吉他 Rainy Blues》[320K/MP3][45.01MB]
- 证声音乐图书馆《雨季 蓝调吉他 Rainy Blues》[FLAC/分轨][109.13MB]
- 赞多《序章》[320K/MP3][45.54MB]
- 许巍.2004-每一刻都是崭新的【步升大风】【WAV+CUE】
- 群星.2024-四方馆影视原声带【韶愔音乐】【FLAC分轨】
- 陈雷.1997-安锁咧【金圆唱片】【WAV+CUE】
- 关淑怡.2013-MY.FAVORITE.SK.3CD【环球】【WAV+CUE】
- Sweety.2006-花言乔语【丰华】【WAV+CUE】