防止 Apache Beam / Dataflow 流 (python) 管道中的融合以消除管道瓶颈

Question

我们目前正在使用 DataflowRunner 在 Apache Beam 上开发流式管道。我们正在从 Pub/Sub 读取消息并对它们进行一些处理，然后我们在滑动窗口中将它们窗口化（当前窗口大小为 3 秒，间隔也是 3 秒）。一旦窗口被触发，我们对窗口内的元素进行一些后处理。这个后处理步骤明显大于窗口大小，大约需要 15 秒。

管道的apache梁代码：

input = ( pipeline | beam.io.ReadFromPubSub(subscription=<subscription_path>)
                   | beam.Map(process_fn))
windows = input | beam.WindowInto(beam.window.SlidingWindows(3, 3),
                                  trigger=AfterCount(30), 
                                  accumulation_mode = AccumulationModel.DISCARDING)
group = windows | beam.GroupByKey()
group | beam.Map(post_processing_fn)

如您所知，Dataflow 会尝试对您的管道步骤执行一些优化。在我们的例子中，它从窗口开始（集群操作：1/处理 2/窗口+后处理）将所有内容融合在一起，这导致只有 1 个工作人员对所有窗口进行缓慢的顺序后处理。我们每 15 秒看到一次管道正在处理下一个窗口的日志。但是，我们希望让多个工作人员选择单独的窗口，而不是将工作量交给单个工作人员。

因此，我们正在寻找防止这种融合发生的方法，因此 Dataflow 将窗口与窗口的后处理分开。通过这种方式，我们希望 Dataflow 能够再次将多个工作人员分配给触发窗口的后处理。

到目前为止我们尝试过的：

• 将工人数量增加到 20、30 甚至 40，但没有效果。仅将窗口化之前的步骤分配给多个工作人员
• 运行管道 5 或 10 分钟，但我们注意到没有工作人员重新分配来帮助窗口后这个更大的后处理步骤
• 窗口化后，将它们放回全局窗口
• 使用虚拟密钥模拟另一个 GroupByKey（如https://cloud.google.com/dataflow/docs/guides/deploying-a-pipeline#preventing-fusion 中所述）但没有任何成功。

最后两个动作确实创建了第三个集群操作（1/ processing 2/ windowing 3/ post-processing ），但我们注意到在窗口化之后仍然是同一个worker在执行所有操作。

是否有任何解决方案可以解决此问题陈述？

我们现在正在考虑的当前解决方法是构建另一个接收窗口的流管道，以便这些工作人员可以并行处理窗口，但它很麻烦。

Answer 1

你已经做了正确的事情来打破你的元素融合。我怀疑可能有问题让您陷入困境。

对于流式传输，单个键总是在同一个 worker 中处理。无论如何，您的所有或大部分记录是否都分配给了一个键？如果是这样，您的处理将在一个工人中完成。

您可以采取的措施是使窗口成为键的一部分，以便多个窗口的元素可以在不同的工作人员中进行处理，即使它们具有相同的键：

class KeyIntoKeyPlusWindow(core.DoFn):
  def process(self, element, window=core.DoFn.WindowParam):
    key, values = element
    yield ((key, window), element)

group = windows | beam.ParDo(KeyIntoKeyPlusWindow() | beam.GroupByKey()

完成后，您可以应用后处理：

group | beam.Map(post_processing_fn)