Airflow 2.8.0中airflow dags test命令未执行任务的原因排查

我使用以下CLI命令运行带参数的Airflow DAG：

airflow dags test first_dag --conf '{"my_parametr":["2023-12-13","ANY"]}'

我的DAG代码结构如下：

def main(**kwargs) -> None:
    my_parametr = kwargs['dag_run'].conf.get('my_parametr')
    print(my_parametr)

task_run_cmd = PythonOperator(task_id='main',
                              python_callable=main,
                              provide_context=True,
                              dag=dag,
                             )

if __name__ == "__main__":
    dag.cli()

补充DAG定义代码：

from airflow import DAG
from airflow.operators.python import PythonOperator

default_args = {
    'owner': 'roman',
    'depends_on_past': False,
    'start_date': datetime(2024, 2, 6),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': 0,
}

dag = DAG(
    'first_dag',
    default_args=default_args,
    description='Your DAG description',
    schedule_interval='0 4 * * *',
)

我预期执行命令后会打印参数["2023-12-13","ANY"]，但仅收到DAG已执行的提示，看不到打印内容，怀疑DAG并未实际运行。使用的Airflow版本为2.8.0，请告知可能的原因。

• 缺少必要导入导致代码报错
  补充的DAG定义代码中未导入datetime模块，但default_args里用到了datetime(2024, 2, 6)，这会直接导致代码运行报错，DAG无法正常加载。需要在代码开头添加：

  from datetime import datetime
  

• provide_context参数已被废弃
  Airflow 2.x版本中，PythonOperator的provide_context=True参数已被废弃，2.8.0版本默认会自动向python_callable传递上下文参数。可以修改PythonOperator定义，移除该废弃参数：

  task_run_cmd = PythonOperator(
      task_id='main',
      python_callable=main,
      dag=dag
  )
  

• print输出未被控制台捕获
  airflow dags test命令默认不会将任务内的print内容直接输出到控制台，你需要查看Airflow日志文件，或者在运行命令时指定DEBUG级别日志来显示输出：

  airflow dags test first_dag --conf '{"my_parametr":["2023-12-13","ANY"]}' -l DEBUG
  

  更稳妥的方式是改用Airflow官方日志模块替代print，确保输出被Airflow日志系统捕获：

  from airflow.utils.log.logging_mixin import LoggingMixin
  
  def main(**kwargs) -> None:
      log = LoggingMixin().log
      my_parametr = kwargs['dag_run'].conf.get('my_parametr')
      log.info(f"my_parametr: {my_parametr}")
  

• DAG未被Airflow正确识别
  确保你的DAG文件放在Airflow的dags_folder目录下（可通过airflow config get core.dags_folder查看路径），且文件名以.py结尾、无语法错误。如果DAG未被加载，airflow dags test不会实际运行任务。

• 执行日期不匹配导致任务跳过
  airflow dags test默认需要指定执行日期，若你的start_date设置为未来日期（比如2024-02-06，当前日期早于该时间），任务会被自动跳过。可以显式指定符合要求的执行日期：

  airflow dags test first_dag 2024-02-06 --conf '{"my_parametr":["2023-12-13","ANY"]}'
  

内容的提问来源于stack exchange，提问作者максим ильин