python-etl框架bonobo使用入门_影子飞扬_etl框架

大大的周 02-07 8121

Bonobo简介

（翻译自https://www.integrate.io/blog/top-etl-python-frameworks） Bonobo自称为“Python3.5+的轻量级提取转换加载（ETL）框架”，包括“用于构建数据转换管道、使用普通Python原语并并行执行它们的工具” 使用Bonobo，开发人员可以轻松地从各种来源提取信息，包括XML/HTML、CSV、JSON、Excel文件和SQL数据库。然后，在将数据加载到目标数据仓库之前，可以使用预构建或自定义转换来应用适当的更改。更具体地说，Bonobo中的数据通过开发人员（即您）定义的Python可调用函数的有向无环图（DAG）中的节点进行传输。只要可能，每个节点都在一个独立的线程上并行运行，大大缩短了运行时间，帮助您避免麻烦的瓶颈。从快速安装过程到用户友好的文档，Bonobo开发人员在构建框架时优先考虑简单性和易用性。Bonobo中ETL管道的构建块是普通的Python对象，Bonobo API尽可能接近基本Python编程语言。Bonobo还包括与许多流行且熟悉的编程工具的集成，如Django、Docker和Jupyter notebooks，以使其更易于启动和运行。一句话：Bonobo是一个ETL Python框架，由于其易用性和许多集成，它适用于许多不同的情况。官网地址：https://www.bonobo-project.org/ 官方文档：http://docs.bonobo-project.org/en/latest/tutorial/1-init.html

安装bonobo

环境：win10，python3.6.8 注意：bonobo在Linux和OSX提供高级支持，而Windows环境则以尽力而为的方式提供支持。这里在win10上做测试。 1、命令窗口执行安装

>pip install bonobo

2、查看安装版本

>bonobo version

3、生成测试实例

>bonobo init my-etl-job.py

4、运行实例

>python tutorial.py Hello World - extract in=1 out=2 [done] - transform in=2 out=2 [done] - load in=2 [done]

5、查看项目流程图（1）安装graphviz 安装指导文档：https://www.graphviz.org/download/ 下载：graphviz-2.50.0 (64-bit) EXE installer [sha256]，并安装。检查环境变量Path中是否添加了Graphviz的命令路径：（2）运行生成流程图命令

>bonobo inspect --graph my-etl-job.py | dot -Tpng -o my-etl-job.png

（3）查看流程图my-etl-job.png

开发简介

1、my-etl-job.py的代码

import bonobo def extract(): """Placeholder, change, rename, remove... """ yield 'hello' yield 'world' def transform(*args): """Placeholder, change, rename, remove... """ yield tuple( map(str.title, args) ) def load(*args): """Placeholder, change, rename, remove... """ print(*args) def get_graph(**options): """ This function builds the graph that needs to be executed. :return: bonobo.Graph """ graph = bonobo.Graph() graph.add_chain(extract, transform, load) return graph def get_services(**options): """ This function builds the services dictionary, which is a simple dict of names-to-implementation used by bonobo for runtime injection. It will be used on top of the defaults provided by bonobo (fs, http, ...). You can override those defaults, or just let the framework define them. You can also define your own services and naming is up to you. :return: dict """ return {} # The __main__ block actually execute the graph. if __name__ == '__main__': parser = bonobo.get_argument_parser() with bonobo.parse_args(parser) as options: bonobo.run( get_graph(**options), services=get_services(**options) )

2、代码说明（1）extract()、transform(*args)、load(*args)方法代表实际进行数据etl过程中的数据获取、转换、加载输出等处理节点，可以根据需要自己定义方法名称及节点个数。可以是原有项目中已经实现的方法。（2）get_graph(**options)方法将数据etl过程中的各个节点，通过graph.add_chain(处理节点方法名列表)组织成有向无环图的处理链。该方法会按照图中节点顺序执行每个节点方法。上一个节点的yield返回值，会作为下一个节点的输入参数传递。而且，每个节点之间的数据是一行一行传递的（见文件尾的注释）。（3）get_services(**options)方法获取所有与外部交互的服务handler，以dict类型返回，比如：注：访问数据库需安装bonobo_sqlalchemy：

>pip install bonobo[sqlalchemy] import sqlalchemy import bonobo_sqlalchemy def get_services(**options): http = requests.Session() http.headers = {'User-Agent': 'Monkeys!'} return { 'fs': bonobo.open_fs('s3://bonobo-examples'),#远程文件系统 'http': http, ## web地址 'sqlalchemy.engine': sqlalchemy.create_engine(...) # sql数据库 } ''' 使用@use获取服务''' @use('http') def extract_fablabs(http): yield from http.get(FABLABS_API_URL).json().get('records') '''使用context.get_service获取服务''' def with_opened_file(self, context): with context.get_service('fs').open('output.txt', 'w+') as f: yield f '''bonobo_sqlalchemy访问数据库''' def get_input(): yield bonobo_sqlalchemy.Select('SELECT * FROM test_in', limit=100)

3、项目结构

Bonobo不强制执行任何类型的项目结构。可以将bonobo的ETL作业文件移动到一个已经存在的项目结构的一个包中，比如mypkg.pipelines。新项目可以使用bonobo自带的项目结构 >bonobo init --package pipelines

注： Bonobo是一种逐行数据流处理解决方案。以这种方式处理数据流会带来以下属性：

先进先出：除非另有说明，否则每个节点都将从FIFO队列接收行，因此，行的顺序将保持不变。每个节点都是这样，但请注意，如果定义“图气泡”（图在不同分支中发散，然后再次收敛），收敛节点将从每个输入队列接收行FIFO，这意味着发散点的顺序在收敛点不会保持为真。并行性：每个节点并行运行（默认情况下，使用独立线程）。这很有用，因为你不必担心阻塞通话。如果线程等待数据库或网络服务，那么其他节点将继续处理数据，只要它们有可用的输入行。独立性：行彼此独立，使得这种处理数据流的方式有利于逐行数据处理，但也不适合“分组”计算（其中输出取决于多行输入数据）。如果需要的输入是相邻行，则可以通过滚动窗口来克服这一点，但是如果需要同时处理整个数据集，则应该考虑其他软件。