使用 bonobo_sqlalchemy (ETL) 在 Bonobo 中进行跨数据库连接答案

【问题标题】：Cross Database Join in Bonobo using bonobo_sqlalchemy (ETL)使用 bonobo_sqlalchemy (ETL) 在 Bonobo 中进行跨数据库连接
【发布时间】：2020-04-14 20:45:16
【问题描述】：

我一直在尝试在 Bonobo 中加入不同的数据源（MySQL 和 PostgreSQL），但到目前为止我一直在努力处理并行性问题，如果 bonobo 是最合适的工具

我知道我可以用这样的方式获取 Bonobo 数据

def get_graph(**options):
    graph = bonobo.Graph()
    graph.add_chain(
        bonobo_sqlalchemy.Select('SELECT * FROM table', limit=100, engine='psql'),
        ...,
    )

    return graph

但在我的情况下，atm 真正需要的是将此获取与另一个数据中的另一个数据连接起来，在另一个表中找到（这可能是 dask/pandas 中的连接）

我知道我可以用这样的方式获取 Bonobo 数据

def get_graph(**options):
    graph = bonobo.Graph()
    graph.add_chain(
        bonobo_sqlalchemy.Select('SELECT * FROM table', limit=100, engine='psql'),
        fetch_mysql_and_join_data,
        ...,
    )

    return graph

在我看来，这有两个问题：

在像 bonobo 这样的并行/分布式提取中加入这样的表是没有意义的；
sqlalchemy 的 bonobo 实现是为使用多处理/线程而构建的；

我怎样才能让它工作？我应该改用其他工具吗？

【问题讨论】：

标签： python etl bonobo-etl

【解决方案1】：

我发现自己在继承的 Bonobo ETL 项目中遇到了类似的问题。对我来说，我想到了两种可能的解决方案：

在链的第二步中，您可以从 MySQL 服务中获取每个相关实体，根据需要“加入”或合并数据，并将其交给下一个节点。
放弃使用bonobo_sqlalchemy 提供的阅读器，将“mysql”和“psql”服务注入到您的初始提取转换中，并自己进行获取/加入。

第一种方法是最容易实现的，但它会产生 N+1 次选择，而且速度可能慢得让人无法忍受。第二种方法可能会更高效，但如果您要处理足够大的数据集，它也可能需要您自己管理数据流。

【讨论】：