使用 BatchElements 进行批处理在 DirectRunner 和 DataflowRunner (GCP/Dataflow) 中的工作方式不同

Question

我正在使用 apache beam (GCP Dataflow) 和 python 构建管道，我的管道如下所示：

...
with beam.Pipeline(options=self.pipeline_options) as pipeline:
            somepipeline = (
                pipeline
                | "ReadPubSubMessage" >> ReadFromPubSub(
                    subscription=self.custom_options.some_subscription)
                | "Windowing" >> beam.WindowInto(beam.window.FixedWindows(30))
                | "DecodePubSubMessage" >> beam.ParDo(DecodePubSubMessage()).with_outputs(ERROR_OUTPUT_NAME, main=MAIN_OUTPUT_NAME)
                | "Geting and sorting listings" >> beam.ParDo(SortByCompletion())
                | "Batching listings" >> beam.BatchElements(min_batch_size=3,max_batch_size=3) 
                | "Print logs" >> beam.Map(logging.info)
            )
...

当我通过 DirectRunner 运行管道时，一切都按预期工作（您可以看到 1 个批次，其中包含 3 个元素）：

但是当我使用 DataflowRunner 运行相同的代码时，我得到了这个结果（您可以看到 3 个批次，每个批次中有 1 个元素）：

即使我并行运行此管道（在两个终端窗口中），也会发生这种情况。两者都使用流式标志运行。消息立即通过python脚本一一发送到pubsub。

问题： 什么会导致 DataflowRunner 出现这个问题（我的假设是数据流中的工作人员数量，但是当我检查它时，这项工作只有 1 个工作人员）以及如何获得与通过 DirrectRunner 相同的结果。

谢谢！

Answer 1

BatchElements 是不确定的，不会跨包进行批处理。直接运行器非常简单，将整个 PCollection 放入一个包中，但 Dataflow 被编写为分布式运行器，即使只有一个工作人员，也可能有多个包同时运行（例如，在不同的线程上），并且包倾向于相当小。

您可以考虑使用 Beam 的 GroupIntoBatches，它在流模式下效果更好（尽管这需要选择一个用于批处理的键）。