Airflow 2.4.1中DAG未按小时运行问题求助

你的核心问题是**schedule_interval参数的位置错误**——你把它放在了default_args字典里，但Airflow中这个参数是属于DAG实例的配置项，而非任务的默认参数。default_args里的参数是传递给DAG内所有任务的，DAG本身不会读取这里的schedule_interval，因此你的DAG实际上使用了错误的调度规则，导致UI显示为每日调度。

修正步骤 #

• 将schedule_interval='@hourly'从default_args中移除，直接作为参数传入DAG的构造函数中。
• Airflow默认catchup=True，只要参数设置正确，会自动补全start_date到当前时间的所有小时级调度实例。

修正后的代码 #

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    'retries': 5,
    'retry_delay': timedelta(minutes=4)
}

with DAG(
    dag_id='dag_with_cron_v6',
    description='dag with cron expression',
    start_date=datetime(2022, 10, 7, 2),
    schedule_interval='@hourly',  # 移至DAG参数中
    default_args=default_args
) as dag:
    task1 = BashOperator(
        task_id='task_1',
        bash_command='echo Please walk carefully. Slippery floor.'
    )

    task1

额外说明 #

• 修正后重启Airflow调度器并刷新UI，调度周期会显示为每小时，catchup逻辑也会正常触发所有遗漏的小时任务实例。
• Airflow 2.x版本中，schedule_interval可与新参数schedule互换使用，你也可以写成schedule='@hourly'，功能完全一致。

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