如何创建每日生成但不自动运行的Airflow DAG？

方案适配场景 #

• 主DAG任务链路：T1 → T2 → T3 → T4 → T5 → T6 → T7
• 非报错场景下因数据变更需重跑指定任务及下游链路（示例为T3→T4→T5）
• 辅助DAG按日生成实例，默认不自动调度，仅支持手动触发指定日期的重跑操作

核心配置要点 #

• 辅助DAG的调度周期、起始日期与主DAG保持一致，确保可生成对应日期的运行实例
• 设置 is_paused_upon_creation=True，保证DAG创建后默认暂停，不会自动触发运行
• 任务逻辑直接调用Airflow CLI清除主DAG指定任务的状态，触发自动重跑

参考代码 #

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

# 替换为你的主DAG ID
MAIN_DAG_ID = "your_main_dag_id"
# 替换为你需要重跑的任务列表
RE_RUN_TASKS = ["T3", "T4", "T5"]

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "start_date": datetime(2024, 1, 1), # 替换为主DAG的起始日期
}

with DAG(
    dag_id="auxiliary_rerun_t3_t5",
    default_args=default_args,
    schedule_interval="@daily", # 替换为主DAG的调度周期
    catchup=True,
    is_paused_upon_creation=True,
    tags=["auxiliary", "rerun"],
) as dag:

    rerun_task = BashOperator(
        task_id="trigger_main_dag_rerun",
        bash_command=f"""
        airflow tasks clear {MAIN_DAG_ID} \
        --task_regex '{'|'.join(RE_RUN_TASKS)}' \
        --execution_date {{ ds }} \
        --yes \
        --rerun_failed
        """,
    )

参数说明 #

• --task_regex：通过正则匹配主DAG中需要重跑的任务，修改RE_RUN_TASKS列表即可调整重跑范围
• --execution_date {{ ds }}：自动取辅助DAG当前运行实例的执行日期，和主DAG对应日期的任务匹配
• --yes：跳过确认提示直接执行清除操作
• --rerun_failed：自动重跑被清除状态的任务，无需额外手动触发

使用方式 #

需要重跑指定日期的T3-T5链路时，在Airflow的DAG列表中找到auxiliary_rerun_t3_t5，选择对应日期的DAG实例，手动触发运行即可。触发后会自动完成主DAG对应任务的状态清除和重跑流程。

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