Google Cloud Composer中Taskflow API定义的DAG无法显示问题咨询

Cloud Composer 完全支持Taskflow API，你的问题大概率是版本兼容性或配置细节问题，以下是具体排查方向：

• 确认Airflow版本：Taskflow API是Airflow 2.0及以上版本的特性，先检查你的Composer环境使用的Airflow版本。如果是1.x版本，直接升级到2.x环境即可解决。

• 检查DAG文件的基础配置

  • 确保DAG文件放在Composer指定的/home/airflow/gcs/dags/根目录下，不要嵌套子文件夹（默认配置不扫描子目录）；
  • 文件名必须以.py结尾，否则Airflow不会识别。

• 验证Taskflow DAG的代码结构
  必须调用被@dag装饰的函数，生成实际的DAG实例，否则Airflow无法检测到。示例正确结构：

  from airflow.decorators import dag, task
  import pendulum
  
  @dag(
      schedule_interval=None,  # 对应你环境的Airflow版本选参数，2.2+可用schedule
      start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
      catchup=False,
      tags=["example"]
  )
  def sample_taskflow_dag():
      @task
      def hello_task():
          return "Hello Taskflow"
      
      hello_task()
  
  # 关键：必须调用函数生成DAG实例
  sample_taskflow_dag()
  

  很多人会遗漏最后一行的函数调用，导致DAG无法被注册。

• 匹配参数与Airflow版本
  你遇到的schedule参数报错，说明你的Composer环境Airflow版本低于2.2.x（schedule是2.2版本才引入的新参数，替代旧的schedule_interval）。请根据环境版本使用对应参数：

  • Airflow <2.2：用schedule_interval=None
  • Airflow >=2.2：推荐用schedule=None

• 查看Scheduler日志排查错误
  如果以上都没问题，去Composer的Airflow UI查看Scheduler日志，或者用gcloud命令拉取日志：

  gcloud composer environments run <你的环境名> logs --location <你的区域> -- scheduler
  

  日志里会显示加载DAG时的具体错误，比如依赖缺失、权限问题等。

• 等待同步时间
  Composer会定期同步GCS上的DAG文件到Airflow实例，一般需要5-10分钟，上传后不要立刻检查，等待一段时间再看UI。

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