控制 Apache Beam Dataflow 管道中的并行性

Question

我们正在尝试使用 Apache Beam（使用 Go SDK）和 Dataflow 来并行化一项耗时的任务。对于更多上下文，我们有缓存作业，它接受一些查询，跨数据库运行它并缓存它们。每个数据库查询可能需要几秒钟到几分钟的时间，我们希望并行运行这些查询以更快地完成任务。

创建了一个简单的管道，如下所示：

    // Create initial PCollection.
    startLoad := beam.Create(s, "InitialLoadToStartPipeline")

    // Emits a unit of work along with query and date range.
    cachePayloads := beam.ParDo(s, &getCachePayloadsFn{Config: config}, startLoad)

    // Emits a cache response which includes errCode, errMsg, time etc.
    cacheResponses := beam.ParDo(s, &cacheQueryDoFn{Config: config}, cachePayloads)

    ...

排放的数量getCachePayloadsFn并不多，生产时大多为数百，最多可达数千。

现在的问题是cacheQueryDoFn不是并行执行，查询是逐个顺序执行的。我们通过在缓存函数中放入日志StartBundle并ProcessElement记录 goroutine id、进程 id、开始和结束时间等来确认这一点，以确认执行中没有重叠。

即使只有 10 个查询，我们也希望始终并行运行查询。根据我们的理解和文档，它根据整体输入创建捆绑包，这些捆绑包并行运行，并且在捆绑包内按顺序运行。有没有办法控制负载中的包数量或增加并行性？

我们尝试过的事情：

1. 保持num_workers=2和autoscaling_algorithm=None. 它启动两个虚拟机，但Setup仅在一个虚拟机上运行初始化 DoFn 的方法，并将其用于整个负载。
2. 在这里找到了sdk_worker_parallelism选项。但不知道如何正确设置。尝试用 来设置它。没有效果。beam.PipelineOptions.Set("sdk_worker_parallelism", "50")

Answer 1

默认情况下，Create 不是并行的，所有 DoFns 都与 Create 融合到同一阶段，因此它们也没有并行性。有关详细信息，请参阅https://beam.apache.org/documentation/runtime/model/#dependent-parallellism 。

您可以使用Reshuffle变换显式强制融合中断。