跨DAG依赖任务未触发：如何实现DAG间任务顺序执行？

核心问题分析 #

你的代码存在语法错误和逻辑配置问题，导致second_dag无法被正常触发且处于暂停状态，具体问题点及修复方案如下：

1. 语法错误修复 #

• Python变量名不能包含空格：Parent DAG中的second task变量名需改为second_task，对应的task_id也需同步修改为second_task（Airflow task_id同样不建议使用空格）。
• 代码缩进与闭合问题：Child DAG中fourth_task的定义未闭合（末尾多了逗号），third_task >> fourth_task的缩进错误，需调整为与任务定义同级。

2. DAG调度与启用配置 #

second_dag作为被触发的DAG，需明确配置：

• 设置schedule_interval=None：避免Airflow按定时调度逻辑处理该DAG。
• 添加is_paused_upon_creation=False：确保DAG创建后自动处于启用状态（也可在Airflow UI手动开启DAG）。

3. 任务触发与传感器匹配逻辑 #

TriggerDagRunOperator触发second_dag时，需传递parent dag的execution_date，让child dag的执行日期与parent保持一致；同时ExternalTaskSensor需配置execution_date_fn来匹配对应的外部任务实例，避免等待错误的任务版本。

修正后的完整代码 #

Parent DAG（first_dag） #

from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.dummy import DummyOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
import logging

SCHEDULE = "59 9 * * 1-5" 
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2025, 2, 26),
    'schedule_interval': SCHEDULE
}

dag = DAG('first_dag', catchup=False, default_args=default_args)

first_task = DummyOperator(
    task_id='first_task',
    dag=dag
)

second_task = TriggerDagRunOperator(
    task_id='second_task',
    trigger_dag_id='second_dag',
    dag=dag,
    # 传递parent dag的执行日期，确保child dag与parent同步
    execution_date="{{ execution_date }}"
)

first_task >> second_task

Child DAG（second_dag） #

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.sensors.external_task import ExternalTaskSensor
import logging

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2025, 2, 26)
}

# 配置为触发式DAG，创建时自动启用
dag = DAG(
    'second_dag', 
    default_args=default_args,
    schedule_interval=None,
    is_paused_upon_creation=False
)

third_task = ExternalTaskSensor(
    task_id='third_task',
    external_dag_id='first_dag',
    external_task_id='second_task',
    # 匹配当前DAG的执行日期对应的外部任务
    execution_date_fn=lambda dt: dt,
    mode='reschedule',  # 减少worker资源占用
    timeout=3600,  # 设置超时时间，防止无限等待
    poke_interval=60  # 每60秒检查一次外部任务状态
)

fourth_task = DummyOperator(
    task_id='fourth_task',
    dag=dag
)

third_task >> fourth_task

额外注意事项 #

• 确保Airflow UI中两个DAG均处于未暂停状态（若代码中未设置is_paused_upon_creation=False，需手动在UI开启second_dag）。
• 若使用Airflow 2.x版本，DummyOperator已移至airflow.operators.dummy模块，需注意导入路径。

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