什么可以限制数据流管道只能使用单个工作人员？

Question

我一直试图调试一个管道，它接受一个驱动后续ParDo操作的输入参数。由于我无法理解的原因，即使我禁用了自动伸缩并设置了工人数量，管道也不会伸缩到单个工人之外。可悲的是，GCP上糟糕的数据流接口几乎没有说明无法伸缩。有谁能告诉你可能是什么问题或者如何有效地调试？

with beam.Pipeline(options=opts) as p:
  result = (
      p | "Initialize Pipeline" >> beam.Create(
          [(f'gs://data/']) |
      "Scan for extraction tasks" >> beam.ParDo(scanner.ScanForTasks()) |
      "Extract data" >> beam.ParDo(worker.TaskToData()))

Answer 1

这个问题被证明与数据流中一种名为“融合”的优化有关，在这种优化中，相邻的操作被融合在一起，大概是为了让它们在同一个工作人员上无缝地运行。问题是，如果一个流水线由一个生成大量下游任务的单个项播种，那么所有这些任务都将在处理初始播种任务的同一个worker上处理。

解决方案是直接在管道中添加任务，以防止这种“优化”降低性能

def scan_for_tasks():
  tasks = []
  # Build your task list here
  return tasks

with beam.Pipeline(options=opts) as p:
  result = (
    p | "Initialize Pipeline" >> beam.Create(scan_for_tasks()) |
    "Extract data" >> beam.ParDo(worker.TaskToData()))