Dataflow Flex模板启动失败：Pickling时发生AttributeError

问题详情 #

我使用Google Cloud Dataflow Flex模板部署Python流式管道，模板无法启动，最终报错：

AttributeError: module 'dill._dill' has no attribute 'log'

此前运行正常，近期dill和beam-sdk均无版本更新，推测是管道序列化出现问题。相关代码及命令如下：

管道代码 #

pipeline_options = PipelineOptions(
    pipeline_args, streaming=True, save_main_session=True, sdk_location="container"
)

with Pipeline(options=pipeline_options) as pipeline:
    (
        pipeline
        | f"Read event topic"
        >> io.ReadFromPubSub(topic=input_topic).with_output_types(bytes)
        | "Convert to string" >> beam.Map(lambda msg: msg.decode("utf=8"))
        | f"Transform event"
        >> beam.Map(transform_message, event_name=event_name)
        | f"Write to output topic"
        >> beam.Map(publish_to_output_topic)
    )

模板构建命令 #

gcloud dataflow flex-template build gs://mybucket/templates/dataflow-latest.json \
    --image "us-docker.pkg.dev/project_id/dataflow/dataflow:latest" \
    --sdk-language "PYTHON"

任务启动命令 #

gcloud dataflow flex-template run "test-job" \
    --template-file-gcs-location "gs://mybucket/templates/dataflow-latest.json" \
    --service-account-email "dataflow@project_id.iam.gserviceaccount.com" \
    --staging-location "gs://mybucket/staging/" \
    --temp-location "gs://mybucket/temp/" \
    --parameters event_name="foobuzz" \
    --parameters sdk_container_image="us-docker.pkg.dev/project_id/dataflow/dataflowsdk:latest" \
    --region "us-central2"

解决方案 #

• 调整save_main_session配置：save_main_session=True会序列化整个主会话状态，可能意外捕获dill内部引用。若管道不依赖全局变量或模块状态，改为save_main_session=False，减少不必要的序列化操作。

• 锁定dill版本：在requirements.txt中明确指定之前验证可用的dill版本（比如dill==0.3.6），避免构建镜像时自动拉取存在兼容问题的patch版本。

• 优化序列化逻辑：

  • 将lambda函数替换为具名函数，避免序列化异常：

    def decode_message(msg):
        return msg.decode("utf-8")
    

    管道中改用beam.Map(decode_message)。
  • 确保transform_message和publish_to_output_topic为纯函数，不引用全局复杂对象，降低序列化复杂度。

• 统一镜像依赖：构建命令和启动命令使用了不同镜像（dataflow:latest vs dataflowsdk:latest），统一使用同一个镜像，避免因镜像内dill版本不一致导致序列化/反序列化不兼容。

• 清理GCS缓存：删除GCS中的模板文件gs://mybucket/templates/dataflow-latest.json和临时目录gs://mybucket/temp/，重新构建并启动模板，排除缓存引发的问题。

内容的提问来源于stack exchange，提问作者Minh