时间:2021-05-22
pyetl是一个纯python开发的ETL框架, 相比sqoop, datax 之类的ETL工具,pyetl可以对每个字段添加udf函数,使得数据转换过程更加灵活,相比专业ETL工具pyetl更轻量,纯python代码操作,更加符合开发人员习惯
安装
pip3 install pyetl
使用示例
数据库表之间数据同步
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()数据库表到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()数据库表同步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()原始表目标表字段名称不同,需要添加字段映射
添加
# 原始表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 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): """通过函数的方式生成字段映射配置,使用更灵活""" # 以下示例将数据库中的字段映射配置取出后转字典类型返回 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
总结
到此这篇关于python ETL工具 pyetl的文章就介绍到这了,更多相关python ETL工具 pyetl内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持!
声明:本页内容来源网络,仅供用户参考;我单位不保证亦不表示资料全面及准确无误,也不保证亦不表示这些资料为最新信息,如因任何原因,本网内容或者用户因倚赖本网内容造成任何损失或损害,我单位将不会负任何法律责任。如涉及版权问题,请提交至online#300.cn邮箱联系删除。
ETL的考虑做数据仓库系统,ETL是关键的一环。说大了,ETL是数据整合解决方案,说小了,就是倒数据的工具。回忆一下工作这么些年来,处理数据迁移、转换的工作倒还
前几天,我去Reddit询问是否应该将Python用于ETL相关的转换,并且压倒性的回答是"是"。但是,尽管我的Redditor同事热心支持使用Python,但
ETL是将业务系统的数据经过抽取、清洗转换之后加载到数据仓库的过程,目的是将企业中的分散、零乱、标准不统一的数据整合到一起,为企业的决策提供分析依据。ETL是B
Sqlmap安装Sqlmap>>基于Python的自动化渗透测试工具,安装工具前需要进行Python的环境准备Python环境的安装1.1下载与安装python
一、python开发工具的选择python开发工具有很多,这里推荐使用pycharm;因为jetbrains公司拥有众多的开发工具以及开发拥护者。这些工具有相同