start_date
和 schedule_interval
参数是否正确。from airflow import DAG
from datetime import datetime, timedelta
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2021, 1, 1),
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}
dag = DAG(
'example_dag',
default_args=default_args,
description='An example DAG',
schedule_interval='*/5 * * * *', # 每 5 分钟触发一次
)
确认调度器(Scheduler)是否在运行。Airflow 调度器分为两种,一种是本地进程调度器(LocalExecutor),另一种是分布式调度器(CeleryExecutor)。如果使用 CeleryExecutor,需要确认 Celery 队列是否启动(参考 Celery 官方文档)。
确认任务运行的执行时间和 DAG 实际运行的时间。在任务详情页中可以查看每次运行的执行时间以及 DAG 上一次运行的时间。
如果仍然存在问题,可以查看调度器日志以及任务详细日志,以确定问题出现的具体位置。