带有 DirectRunner 的 Apache Beam (SUBPROCESS_SDK) 仅使用一个工作人员，我如何强制它使用所有可用的工作人员？

Posted 2023-02-16

【中文标题】带有 DirectRunner 的 Apache Beam (SUBPROCESS_SDK) 仅使用一个工作人员，我如何强制它使用所有可用的工作人员？

【英文标题】：Apache Beam with DirectRunner (SUBPROCESS_SDK) uses only one worker, how do I force it to use all available workers?

【发布时间】：2020-03-01 03:08:58

【问题描述】：

以下代码：

def get_pipeline(workers):
    pipeline_options = PipelineOptions(['--direct_num_workers', str(workers)])
    return beam.Pipeline(options=pipeline_options,
                         runner=fn_api_runner.FnApiRunner(
                             default_environment=beam_runner_api_pb2.Environment(
                                 urn=python_urns.SUBPROCESS_SDK,
                                 payload=b'%s -m apache_beam.runners.worker.sdk_worker_main'
                                         % sys.executable.encode('ascii'))))

with get_pipeline(4) as pipeline:
  _ = (  
        pipeline
        | 'ReadTestData' >> beam.io.ReadFromParquet(input_files, columns=all_columns)
        | "write" >> beam.io.WriteToText("/tmp/txt2")
  )

只使用 4 个可用的 worker 中的一个，并且只生成一个大的输出文件（即使有很多输入文件）。

如何强制 Beam 管道并行工作，即如何强制每个输入文件由不同的工作人员单独处理？

【问题讨论】：

我也有同样的问题。在查看 htop 和 Beam 日志时似乎只使用一个 CPU 内核。

【参考方案1】：

您使用的是哪个版本的光束？

我对梁 2.16.0 有同样的问题，但版本 2.17.0 似乎有预期的行为。

您可能想尝试使用此版本，同时保持您的代码不变。