如何配置Airflow DAG在上次触发仍运行时跳过当小时调度

原配置存在一处写法错误：schedule_interval是DAG初始化的独立入参，写在default_args中不会生效。
要实现需求无需额外开发判断逻辑，通过两个DAG核心参数即可实现：

核心参数作用 #

• max_active_runs=1：DAG维度并发限制，配置后同一个DAG同一时间最多只会有1个处于运行状态的DAG实例，从根源避免多周期任务同时执行。
• catchup=False：关闭历史调度补跑逻辑。到达新的调度时间点时，如果检测到该DAG已有活跃运行实例，调度器不会为当前周期创建排队等待的实例，直接跳过本次调度，等到下一个调度周期再检查运行条件。

可直接复用的配置示例 #

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.utils import dates
from datetime import timedelta

default_args = {
    'owner': 'Airflow',
    'start_date': dates.days_ago(1),
    'email': ['email1@gmail.com', 'email2@gmail.com'],
    'email_on_failure': True,
    'email_on_success': True,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    dag_id='hourly_skip_dag',
    default_args=default_args,
    schedule_interval='0 * * * *', # 调度规则必须写在DAG初始化参数中，不能放在default_args里
    max_active_runs=1,
    catchup=False
) as dag:
    # 替换为实际业务任务
    start = DummyOperator(task_id='start')
    end = DummyOperator(task_id='end')

    start >> end

避坑提示 #

• 不要给任务配置depends_on_past=True。该参数是任务级依赖配置，开启后会要求当前任务的上一周期实例必须执行成功才能触发当前实例，会导致后续调度排队卡住，不符合跳过当前调度的需求，保持默认的False即可。
• 如果使用Airflow 1.x版本，可额外添加DAG参数concurrency=1，限制DAG内同时运行的任务实例数为1，进一步规避并发问题；Airflow 2.x版本中该参数已更名为max_active_tasks_per_dag，默认配置即可满足需求，无需额外调整。

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