使用自定义Artifact容器运行Dataflow任务时卡住崩溃求助

核心问题分析 #

任务卡住无worker活动，通常是自定义容器配置不兼容、依赖版本不匹配、权限缺失或Pipeline参数错误导致的。以下是针对性修复步骤：

1. 修复Dockerfile版本不兼容问题 #

基础镜像用了Python 3.11，但复制的是apache/beam_python3.10_sdk:2.51.0的内容，Python版本不匹配会导致SDK运行异常。同时，Dataflow自定义容器无需指定CMD，boot进程会自动处理Pipeline执行逻辑。

修改后的Dockerfile：

# 使用与SDK镜像一致的Python版本
FROM python:3.10-slim

WORKDIR /app

COPY . /app

# 指定与SDK镜像严格一致的apache-beam版本，避免依赖冲突
RUN pip install --no-cache-dir apache-beam[gcp]==2.51.0 google-cloud-storage

RUN pip check

# 复制对应版本的SDK内容
COPY --from=apache/beam_python3.10_sdk:2.51.0 /opt/apache/beam /opt/apache/beam

ENTRYPOINT ["/opt/apache/beam/boot"]

2. 修正requirements.txt版本匹配 #

确保apache-beam版本与SDK镜像版本完全一致：

apache-beam[gcp]==2.51.0
google-cloud-storage

3. 修复Pipeline运行参数错误 #

• Windows命令提示符换行需用^而非\，否则参数会被截断
• output_file必须是GCS路径（如gs://your-bucket/output/results），不能是本地路径
• 必须指定临时文件目录--temp_location，Dataflow依赖该目录处理中间数据
• 参数需正确分隔，避免遗漏

修正后的运行命令（Windows CMD）：

python dataflow_job_script.py ^
    --region us-central1 ^
    --runner DataflowRunner ^
    --project PROJECT-NAME ^
    --sdk_container_image us-central1-docker.pkg.dev/PROJECT-NAME/list-objects-artifact-repo-v4/list-objects-docker-image-v2 ^
    --sdk_location=container ^
    --temp_location=gs://YOUR_BUCKET_NAME/temp ^
    --bucket YOUR_BUCKET_NAME ^
    --output gs://YOUR_BUCKET_NAME/output/results

同时修改dataflow_job_script.py，改为从命令行参数读取配置，避免硬编码：

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

class ListGCSObjects(beam.DoFn):
    def __init__(self, bucket):
        self.bucket = bucket

    def process(self, element):
        from google.cloud import storage
        client = storage.Client()
        bucket = client.bucket(self.bucket)

        for blob in bucket.list_blobs():
            yield blob.name

def run_pipeline():
    options = PipelineOptions()
    config = options.get_all_options()
    bucket_name = config.get('bucket')
    output_file = config.get('output')

    with beam.Pipeline(options=options) as p:
        (
            p
            | 'Create' >> beam.Create([None])
            | 'List GCS Objects' >> beam.ParDo(ListGCSObjects(bucket_name))
            | 'Write to File' >> beam.io.WriteToText(output_file)
        )

if __name__ == '__main__':
    run_pipeline()

4. 验证Worker权限 #

确保Dataflow Worker使用的服务账号（默认是PROJECT-NAME@appspot.gserviceaccount.com）拥有以下权限：

• storage.objects.list、storage.objects.create（GCS读写权限）
• artifacts.repositories.downloadArtifacts（Artifact Registry镜像拉取权限）
  可通过IAM控制台为服务账号添加Dataflow Worker、Storage Object Admin和Artifact Registry Reader角色。

5. 检查Worker日志 #

通过Cloud Logging筛选Dataflow任务的worker日志，搜索ERROR或Exception关键词定位具体启动失败原因，筛选条件：

resource.type="dataflow_step" AND labels.dataflow_job_id="YOUR_JOB_ID"

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