python - Apache Beam Pipeline 与 DirectRunner 一起运行，但在初始读取步骤期间因 DataflowRunner(SDK 线束 sdk-0-0 断开连接)而失败

Question

长话短说

我们有一个默认的 VPC。试图运行数据流作业。初始步骤(读取文件)设法处理 1/2 个步骤。获取 JOB_MESSAGE_ERROR: SDK harness sdk-0-0 disconnected 错误消息，但日志中没有其他内容。已尝试设置角色和 vpc 防火墙规则。

问题

我想使用 Geobeam 图像(Apache Beam Python 3.9 SDK 2.41.0)运行数据流作业。我已将作业定义如下:

def run(pipeline_args, known_args):
    import apache_beam as beam
    from apache_beam.io.gcp.internal.clients import storage
    from apache_beam.options.pipeline_options import PipelineOptions
    from geobeam.io import GeoJSONSource, filebasedsource
    from geobeam.fn import format_record, make_valid, filter_invalid

    pipeline_options = PipelineOptions([
        
    ] + pipeline_args)

    with beam.Pipeline(options=pipeline_options) as p:
        (p
         | beam.io.Read(GeoJSONSource(known_args.gcs_url, encoding='utf-8'))
         | 'FilterCords' >> beam.Filter(lambda x: len(x[-1]["coordinates"]) > 1)
         | 'MakeValid' >> beam.Map(make_valid)
         | 'FilterInvalid' >> beam.Filter(filter_invalid)
         | 'FormatRecords' >> beam.Map(format_record)
         | beam.io.WriteToText(known_args.gcs_write_url)
        )

if __name__ == '__main__':
    import logging
    import argparse

    logging.getLogger().setLevel(logging.INFO)

    parser = argparse.ArgumentParser()
    parser.add_argument('--gcs_url')
    parser.add_argument('--gcs_write_url')
    known_args, pipeline_args = parser.parse_known_args()

    run(pipeline_args, known_args)

我使用以下命令运行作业:

python -m main --runner DataflowRunner --project [[project_id]] \
--temp_location gs://[[temp_bucket_name]]/tmp \
--gcs_url gs://[[inputbucket_name]]/[[filename]].geojson \
--region europe-north1  --sdk_container_image gcr.io/dataflow-geobeam/example \
--gcs_write_url gs://gs://[[outputbucket_name]]/[[filename]]_processed.geojson \
--subnetwork [[full_link_to_subnet]]

我们已经设置了自定义默认 VPC，并且我在 GCP 中为计算虚拟机资源添加了入口/导出防火墙规则的推荐范围。我还为用于数据流作业的默认服务帐户赋予了以下角色:

• 计算网络用户
• 数据流管理员
• 数据流 worker
• 存储对象管理员

我还在服务帐户上提供了我的用户角色:

• 所有者
• 服务帐户管理员

管道输出

它说作业已停止，但那是因为作业不会继续进行。我得到以下日志输出

INFO:apache_beam.runners.dataflow.dataflow_runner:Job 2022-10-18_05_33_31-17288646308046950877 is in state JOB_STATE_PENDING
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:31.708Z: JOB_MESSAGE_BASIC: Dataflow Runner V2 auto-enabled. Use --experiments=disable_runner_v2 to opt out.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:32.780Z: JOB_MESSAGE_DETAILED: Autoscaling is enabled for job 2022-10-18_05_33_31-17288646308046950877. The number of workers will be between 1 and 1000.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:32.803Z: JOB_MESSAGE_DETAILED: Autoscaling was automatically enabled for job 2022-10-18_05_33_31-17288646308046950877.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:34.374Z: JOB_MESSAGE_BASIC: Worker configuration: n1-standard-1 in europe-north1-b.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.092Z: JOB_MESSAGE_DETAILED: Expanding SplittableParDo operations into optimizable parts.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.109Z: JOB_MESSAGE_DETAILED: Expanding CollectionToSingleton operations into optimizable parts.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.141Z: JOB_MESSAGE_DETAILED: Expanding CoGroupByKey operations into optimizable parts.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.160Z: JOB_MESSAGE_DEBUG: Combiner lifting skipped for step WriteToText/Write/WriteImpl/GroupByKey: GroupByKey not followed by a combiner.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.184Z: JOB_MESSAGE_DETAILED: Expanding GroupByKey operations into optimizable parts.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.200Z: JOB_MESSAGE_DEBUG: Annotating graph with Autotuner information.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.226Z: JOB_MESSAGE_DETAILED: Fusing adjacent ParDo, Read, Write, and Flatten operations
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.243Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/InitializeWrite into WriteToText/Write/WriteImpl/DoOnce/Map(decode)
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.262Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/DoOnce/FlatMap(<lambda at core.py:3481>) into WriteToText/Write/WriteImpl/DoOnce/Impulse
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.278Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/DoOnce/Map(decode) into WriteToText/Write/WriteImpl/DoOnce/FlatMap(<lambda at core.py:3481>)
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.294Z: JOB_MESSAGE_DETAILED: Fusing consumer Read/Map(<lambda at iobase.py:908>) into Read/Impulse
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.310Z: JOB_MESSAGE_DETAILED: Fusing consumer ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/PairWithRestriction into Read/Map(<lambda at iobase.py:908>)
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.325Z: JOB_MESSAGE_DETAILED: Fusing consumer ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/SplitWithSizing into ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/PairWithRestriction
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.340Z: JOB_MESSAGE_DETAILED: Fusing consumer FilterCords into ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/ProcessElementAndRestrictionWithSizing
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.356Z: JOB_MESSAGE_DETAILED: Fusing consumer MakeValid into FilterCords
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.372Z: JOB_MESSAGE_DETAILED: Fusing consumer FilterInvalid into MakeValid
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.387Z: JOB_MESSAGE_DETAILED: Fusing consumer FormatRecords into FilterInvalid
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.402Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/WindowInto(WindowIntoFn) into FormatRecords
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.417Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/WriteBundles into WriteToText/Write/WriteImpl/WindowInto(WindowIntoFn)
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.432Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/Pair into WriteToText/Write/WriteImpl/WriteBundles
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.447Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/GroupByKey/Write into WriteToText/Write/WriteImpl/Pair
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.464Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/Extract into WriteToText/Write/WriteImpl/GroupByKey/Read
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.489Z: JOB_MESSAGE_DEBUG: Workflow config is missing a default resource spec.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.504Z: JOB_MESSAGE_DEBUG: Adding StepResource setup and teardown to workflow graph.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.519Z: JOB_MESSAGE_DEBUG: Adding workflow start and stop steps.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.535Z: JOB_MESSAGE_DEBUG: Assigning stage ids.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.624Z: JOB_MESSAGE_DEBUG: Executing wait step start19
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.655Z: JOB_MESSAGE_BASIC: Executing operation Read/Impulse+Read/Map(<lambda at iobase.py:908>)+ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/PairWithRestriction+ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/SplitWithSizing
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.668Z: JOB_MESSAGE_BASIC: Executing operation WriteToText/Write/WriteImpl/DoOnce/Impulse+WriteToText/Write/WriteImpl/DoOnce/FlatMap(<lambda at core.py:3481>)+WriteToText/Write/WriteImpl/DoOnce/Map(decode)+WriteToText/Write/WriteImpl/InitializeWrite
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.682Z: JOB_MESSAGE_DEBUG: Starting worker pool setup.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.696Z: JOB_MESSAGE_BASIC: Starting 1 workers in europe-north1-b...
INFO:apache_beam.runners.dataflow.dataflow_runner:Job 2022-10-18_05_33_31-17288646308046950877 is in state JOB_STATE_RUNNING
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:34:21.585Z: JOB_MESSAGE_DETAILED: Autoscaling: Raised the number of workers to 1 based on the rate of progress in the currently running stage(s).
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:37:30.456Z: JOB_MESSAGE_DETAILED: Workers have started successfully.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:42:40.315Z: JOB_MESSAGE_BASIC: Finished operation Read/Impulse+Read/Map(<lambda at iobase.py:908>)+ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/PairWithRestriction+ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/SplitWithSizing
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:42:40.354Z: JOB_MESSAGE_DEBUG: Value "ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6-split-with-sizing-out3" materialized.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:42:42.422Z: JOB_MESSAGE_ERROR: SDK harness sdk-0-0 disconnected.

然后它再次尝试将 worker 数量增加到 1，然后它立即收到 JOB_MESSAGE_ERROR: SDK harness sdk-0-0 disconnected. 一遍又一遍。附带说明 - 在管道实际启动之前还需要大约 10 分钟。

帮助

我设法让它与 DirectRunner 选项一起工作。我不知道去哪里看？会不会跟VPC有关？

编辑:它可能是 Geobeam 图像吗？

我尝试在原生/默认图像和地理束图像上运行字数统计示例，它适用于原生/默认图像，但不适用于地理束图像。

为什么会这样？

Answer 1

经过反复试验，我发现geobeam 基础镜像的python 版本必须与您机器上的本地python 版本相匹配，否则将无法运行。在回答时，这是 python 3.8。