google-cloud-dataflow - 使用 Apache Beam 从数据库中读取批量数据

Question

我想知道，如果我的查询返回数百万行，JdbcIO 将如何并行执行查询。 我提到了https://issues.apache.org/jira/browse/BEAM-2803以及相关的拉取请求。我完全听不懂。

ReadAll expand 方法使用了 ParDo。因此它会创建多个数据库连接来并行读取数据吗？如果我限制可以创建到数据源中的数据库的连接数，它会坚持连接限制吗？

任何人都可以帮助我理解这将如何在 JdbcIO 中处理吗？我正在使用 2.2.0

更新:

.apply(
          ParDo.of(
              new ReadFn<>(
                  getDataSourceConfiguration(),
                  getQuery(),
                  getParameterSetter(),
                  getRowMapper())))

以上代码显示 ReadFn 与 ParDo 一起应用。我认为，ReadFn 将并行运行。如果我的假设是正确的，我将如何使用 readAll() 方法从我一次只能建立有限数量的连接的数据库中读取数据？

谢谢 巴鲁

Answer 1

ReadAll 方法处理您有多个查询的情况。您可以将查询存储为字符串的 PCollection，其中每个字符串都是查询。然后在读取时，每个项目都在单个 ParDo 中作为单独的查询处理。

这不适用于少量查询，因为它将并行性限制为查询数量。但如果你有很多，那么它会执行得更快。大多数 ReadAll 调用都是这种情况。

从代码看来，连接是在设置函数中为每个工作人员建立的。这可能包括多个查询，具体取决于工作人员的数量和查询的数量。

在哪里设置查询限制？无论有没有 ReadAll，它的行为都应该相似。

有关更多信息，请参阅 jira:https://issues.apache.org/jira/browse/BEAM-2706

我对 jdbcIO 不是很熟悉，但他们似乎实现了 jira 中建议的版本。 PCollection 可以是任何东西，然后回调以根据 PCollection 中的元素修改查询。这允许 PCollection 中的每个项目代表一个查询，但比将新查询作为每个元素更灵活一些。