Apache Airflow（MWAA）托管工作流如何禁用任务对前次运行的依赖

架构老林

2026-9-30

问题根因

你遇到的调度阻塞问题和depends_on_past参数无关，该参数仅控制单个任务是否需要等待上一次调度的同任务执行成功，不影响DAG层面的调度逻辑，常见触发原因有4类：

DAG级max_active_runs配置为默认值1，且存在状态异常的DAG Run残留。Airflow默认限制单个DAG同时最多运行1个实例，如果你前一次失败的DAG Run因为元数据异常被统计为活跃实例，后续生成的DAG Run会因为配额不足无法调度，状态显示为None。如果你的DAG开启了catchup=True，还会生成大量积压的历史调度实例，进一步占用配额。
触发一级DAG的任务配置不合理。你二级DAG的单任务大概率用的是TriggerDagRunOperator且开了wait_for_completion=True，如果上一次触发的一级/零级DAG因为ECS任务卡住、状态上报异常等问题没有返回终态，会导致二级DAG的任务一直处于运行状态，占用max_active_runs配额。
单DAG并发数限制。你DAG级的concurrency参数如果设为1，且上一次失败的任务实例存在状态残留，新DAG Run中的任务会因为单DAG任务并发数不足无法被调度。
托管环境自定义规则限制。部分Airflow托管服务（如AWS MWAA、GCP Cloud Composer）内置了失败DAG调度拦截逻辑，或者你部署的自定义调度插件存在拦截规则，会在DAG连续失败后暂停后续调度。

排查解决步骤

优先调整二级DAG的核心配置，显式指定相关参数避免默认值限制，参考配置如下：

from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.trigger_dagrun import TriggerDagRunOperator

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "retries": 1,
    "execution_timeout": timedelta(minutes=12) # 单任务最长运行12分钟强制超时
}

with DAG(
    dag_id="your_level2_dag_id",
    schedule_interval="*/15 * * * *",
    default_args=default_args,
    max_active_runs=3, # 允许同时最多3个DAG实例运行
    concurrency=3, # 单DAG允许同时最多运行3个任务
    catchup=False, # 关闭历史补跑，避免生成积压调度实例
    start_date=datetime(2024,1,1)
) as dag:
    trigger_level1 = TriggerDagRunOperator(
        task_id="trigger_level1_dag",
        trigger_dag_id="your_level1_dag_id",
        wait_for_completion=True,
        poke_interval=30, # 每30秒拉取一次一级DAG运行状态
        allowed_states=["success"],
        failed_states=["failed", "upstream_failed", "skipped"]
    )

清理异常状态的DAG Run和任务实例。进入Airflow UI的Browse菜单，分别查看DAG Runs和Task Instances页面，筛选对应二级DAG的记录，手动标记状态异常的running实例为failed，释放调度配额。
检查托管环境的调度规则。如果使用云厂商托管的Airflow服务，确认没有开启失败DAG自动暂停调度的相关配置，排查是否有自定义插件拦截了调度逻辑。
如果使用的是2.4版本以下的Airflow，建议升级到2.6+稳定版，低版本Airflow存在max_active_runs计数异常的已知bug，失败的DAG Run会被错误统计为活跃实例阻塞后续调度。

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