GCP Dataflow无法读取自行写入的pipeline.pb及后续报错求助

原始问题 #

运行Dataflow Pipeline的命令：

!python3 ~/pipelines/Beam/pipeline.py \
  --project='project_id' \
  --region='region' \
  --dataset_id='dataset_id' \
  --ingest_table_name='titanic_train' \
  --bucket='bucket_id' \
  --temp_location='gs://bucket_id/dataflow/temp' \
  --runner=DataflowRunner

现象与验证 #

• Pipeline成功将pipeline.pb写入指定temp_location（路径示例：gs://bucket_id/dataflow/temp/my-bq-pipeline-...../my-bq-pipeline-..../pipeline.pb）
• 后续读取该文件时抛出错误：

INFO:apache_beam.runners.dataflow.dataflow_runner: JOB_MESSAGE_ERROR: Unable to open file: gs://bucketId/dataflow/temp/my-bq-pipeline-..../my-bq-pipeline-..../pipeline.pb.

• 已验证：通过Notebook执行gsutil cp gs://bucket_id/dataflow/temp/my-bq-pipeline-..../my-bq-pipeline-..../pipeline.pb ./pipeline.pb可正常访问该文件；使用DirectRunner运行Pipeline无异常。

更新1：设置service_account_email后出现作业重复错误 #

添加service_account_email参数后触发新错误：

"apache_beam.runners.dataflow.internal.apiclient.DataflowJobAlreadyExistsError: There is already active job named my-bq-pipeline-169... with id: 2023-10-.... If you want to submit a second job, try again by setting a different name using --job_name"

确认：Dataflow作业列表中无同名的第二个作业，初始作业名称为唯一值。

更新2：强制生成唯一作业名称仍报错 #

代码中通过以下逻辑生成唯一作业名称：

options.view_as(GoogleCloudOptions).job_name = '{0}{1}'.format('my-bq-pipeline-123-',time.time_ns())

已修改前缀确保唯一性，但执行时仍触发相同的作业重复错误，且gcloud dataflow jobs list输出中无重复job_name。

解决方案 #

针对pipeline.pb读取失败问题 #

• 检查服务账号权限：确认Dataflow使用的服务账号（默认是[项目编号]-compute@developer.gserviceaccount.com，指定service_account_email则为该账号）拥有GCS存储桶的storage.objects.get权限，可通过IAM控制台添加Storage Object Viewer角色。注意：Notebook的执行身份与Dataflow服务账号可能不同，前者有权限不代表后者也有。
• 校验路径大小写：GCS路径大小写敏感，错误信息中路径为gs://bucketId/...，实际路径是gs://bucket_id/...，需确保命令及代码中引用的路径大小写完全一致。
• 排查代码路径逻辑：检查代码中读取pipeline.pb的部分，确认路径拼接无硬编码、变量替换错误等问题。

针对作业重复名称错误 #

• 检查名称长度：Dataflow作业名称最长为100字符，若代码中存在额外拼接逻辑，可能导致名称截断或重复，需确保最终生成的名称符合长度要求且唯一。
• 清除本地缓存：清理Beam本地缓存目录（默认~/.apache_beam/cache），或重启Notebook内核后重新提交作业，避免本地缓存的旧作业信息干扰。
• 确认区域一致性：提交作业时的--region参数需与gcloud dataflow jobs list查询的区域一致，否则可能漏看跨区域的作业。
• 命令行指定job_name：跳过代码中的名称设置逻辑，直接在运行命令中添加--job_name=my-bq-pipeline-$(date +%s%N)，强制生成唯一名称。
• 升级Beam SDK并清除客户端缓存：升级到最新版Beam SDK，或在代码中添加以下代码清除客户端缓存：

  from apache_beam.runners.dataflow.dataflow_runner import DataflowRunner
  DataflowRunner._cached_job = None
  

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