使用Airflow BeamRunPythonPipelineOperator运行Python流水线失败求助

你遇到的核心问题是word_count.py中硬编码覆盖了入参argv，导致Airflow Operator传递的运行参数全部失效，同时你硬写了--template_location参数，该参数会让流水线仅生成模板文件，不会实际提交Dataflow作业运行，这就是你看不到Dataflow作业、也没有结果文件的根本原因。

1. 修复word_count.py的入参逻辑 #

删除你在run函数中硬写argv的代码段，保留Airflow传递的入参即可，修改后的核心代码如下：

def run(argv=None, save_main_session=True):
  """Main entry point; defines and runs the wordcount pipeline."""
  parser = argparse.ArgumentParser()
  parser.add_argument(
      '--input',
      dest='input',
      default='gs://<...>/kinglear.txt',
      help='Input file to process.')
  parser.add_argument(
      '--output',
      dest='output',
      default='gs://<...>/output.txt',
      help='Output file to write results to.')

  # 删掉原来你硬编码赋值argv的那段代码，直接使用传入的argv参数
  known_args, pipeline_args = parser.parse_known_args(argv)

  pipeline_options = PipelineOptions(pipeline_args)
  pipeline_options.view_as(SetupOptions).save_main_session = save_main_session

  # 这里创建Pipeline不需要再传argv，直接用解析后的pipeline_options即可
  with beam.Pipeline(options=pipeline_options) as p:
    # 后续的流水线逻辑（读取文件、分词、计数、写结果）保持原有代码不变

2. DAG配置检查项 #

• 确认DAG中填写的所有GCS路径（py_file、输入输出路径、临时存储路径）都真实存在，且Composer服务账号有对应路径的读写权限
• 确认DataflowConfiguration中的project_id、location和你GCP项目的实际配置一致
• 如果你的Composer环境内置的apache-beam版本和你指定的py_requirements中的2.26.0差异过大，建议对齐版本避免兼容问题

• 优先查看Airflow任务日志：在Composer的DAG页面点击对应任务的「查看日志」按钮，就能看到具体的报错信息，比如权限不足、路径不存在、参数错误等具体原因
• 确认服务账号权限：Composer使用的默认服务账号需要具备Dataflow开发者、存储对象管理员权限，否则作业无法正常提交
• 本地验证流水线：你可以在本地配置好GCP认证后，直接运行word_count.py测试流水线是否能正常提交Dataflow作业，确认流水线本身无问题后再放到Airflow上调度
• 移除模板相关参数：如果你不需要预生成Dataflow模板，不要在任何位置添加--template_location参数，该参数会跳过作业运行步骤

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