Airflow调用DataFlowPythonOperator执行任务失败返回码2求助

我来帮你捋一捋这个问题，本地能正常跑但Airflow调度DataFlow作业失败，这种情况大多是权限或者环境配置的差异导致的，给你几个实用的排查和解决方向：

1. 先确认Airflow所用服务账号的权限 #

你本地运行脚本时用的应该是个人GCP账号权限，但Airflow是通过指定的服务账号来执行DataFlow作业的，这两者权限可能不一样：

• 给Airflow的服务账号至少添加roles/dataflow.developer（DataFlow作业提交权限）和roles/storage.objectViewer（GCS文件读取权限），如果作业还有写入操作，要对应加上写入权限。
• 可以去GCP IAM控制台直接给这个服务账号分配上述角色，避免因权限不足导致作业启动失败。

2. 检查DataFlowPythonOperator的参数配置 #

本地运行时gcloud会自动读取本地配置的项目、区域，但Airflow里必须显式指定关键参数，不然容易出错：

• 确保你的Operator定义里明确指定了project_id、region，还有GCS上的脚本路径和输入路径：

  run_dataflow = DataFlowPythonOperator(
      task_id="execute_dataflow",
      py_file="gs://your-bucket/path/to/your_script.py",
      project_id="your-gcp-project-id",
      region="us-central1",  # 改成你实际用的区域
      gcp_conn_id="google_cloud_default",
      options={
          "input": "gs://your-input-bucket/*.txt",
          # 其他DataFlow作业需要的参数
      }
  )
  

• 注意py_file如果是本地路径，要保证所有Airflow Worker节点都能访问到；更稳妥的方式是把脚本上传到GCS，用GCS路径。

3. 验证Airflow的GCP连接有效性 #

即使你用了默认连接，也有可能连接配置有问题：

• 打开Airflow UI，进入「Admin」→「Connections」，找到google_cloud_default，检查服务账号密钥JSON是否正确上传，或者如果用的是工作负载身份，确认服务账号邮箱配置正确。
• 可以先加一个简单的测试任务，比如用GCSSensor检查目标GCS文件是否存在，验证这个连接能不能正常访问GCP资源，排除连接本身的问题。

4. 查看DataFlow作业的详细报错日志 #

Airflow里的日志只显示了失败的简略信息，具体的报错原因得去GCP Console看：

• 登录GCP Console，进入DataFlow页面，找到这个失败的作业，查看「日志」标签页，特别是Worker节点的报错信息——比如有没有PermissionDenied的提示，或者输入文件路径错误、依赖缺失等细节。

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