如何基于列表在Airflow中创建带外部DAG依赖的任务工作流

核心问题 #

你的代码存在两个关键问题导致wait_task一直处于运行状态：

1. 任务依赖逻辑混乱：循环中重复将trigger_operator指向不同的previous_task，导致依赖链断裂，外部DAG未与对应任务正确绑定。
2. ExternalTaskSensor参数配置错误：未匹配外部DAG的实际执行日期，传感器一直在等待不存在的执行实例；同时你指定了单个external_task_id，但需求是等待外部DAG所有任务完成，参数设置不符合预期。

修正步骤与代码实现 #

1. 梳理正确执行流程 #

按照需求，流程应按以下顺序执行：
触发外部DAG → 等待外部DAG全部完成 → 执行当前DAG任务 → 触发下一个外部DAG → ...

2. 修正ExternalTaskSensor参数 #

• 设置external_task_id=None：表示等待外部DAG的所有任务完成，而非单个任务。
• 添加execution_date_fn：匹配外部DAG的实际执行日期（避免因触发方式导致的日期不匹配问题）。
• 配置poke_interval和timeout：控制传感器检查频率与超时时间，避免无限等待。

修正后的完整代码 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.sensors.external_task import ExternalTaskSensor
from datetime import datetime

def get_external_execution_date(context):
    # 获取当前DAG的执行日期，若外部DAG有特殊触发逻辑，可按需调整
    return context['execution_date']

dag = DAG(
    "MY_DAG",
    start_date=datetime(2023, 1, 1),
    schedule="@daily",
    catchup=False
)

def ex_func_airflow(i):
    print(f"执行任务task_tab_{i}")

tabs = [1, 2, 3]
previous_task = None

for i in tabs:
    # 触发外部DAG的任务
    trigger_task = TriggerDagRunOperator(
        task_id=f'trigger_external_dag_{i}',
        trigger_dag_id='EXTERNAL_DAG_ID',  # 替换为你的外部DAG真实ID
        dag=dag
    )
    
    # 等待外部DAG所有任务完成的传感器
    wait_task = ExternalTaskSensor(
        task_id=f'wait_external_dag_{i}',
        external_dag_id='EXTERNAL_DAG_ID',
        external_task_id=None,  # 设为None表示等待整个外部DAG完成
        execution_date_fn=get_external_execution_date,
        poke_interval=30,  # 每30秒检查一次外部DAG状态
        timeout=3600,  # 超时1小时后终止等待
        dag=dag
    )
    
    # 当前DAG的业务任务
    current_task = PythonOperator(
        task_id=f'task_tab_{i}',
        op_args=[i],
        python_callable=ex_func_airflow,
        dag=dag
    )
    
    # 构建依赖链
    if previous_task:
        previous_task >> trigger_task >> wait_task >> current_task
    else:
        # 第一个任务直接从触发外部DAG开始
        trigger_task >> wait_task >> current_task
    
    previous_task = current_task

额外检查项 #

• 确认外部DAG的IDEXTERNAL_DAG_ID拼写完全正确。
• 确保外部DAG能正常执行完成，若外部DAG失败或卡住，传感器会持续等待。
• 若外部DAG需多次触发且需区分实例，可在TriggerDagRunOperator中添加conf参数传递标识，在get_external_execution_date中根据conf匹配对应外部DAG实例。

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