如何使用单个ExternalTaskSensor等待多个外部任务执行完成？

你完全不需要维护两个独立的ExternalTaskSensor，Airflow的ExternalTaskSensor本身就支持传入任务ID列表，来等待多个外部任务全部完成后再触发后续任务。

核心实现思路 #

关键就是利用ExternalTaskSensor的external_task_ids参数——这个参数接受一个字符串列表，传感器会持续检查，直到列表中所有指定的任务都进入你预设的成功状态，才会结束等待。

代码示例 #

下面是一个具体的实现样例，对应你的业务场景：

from airflow.sensors.external_task import ExternalTaskSensor
from airflow.operators.dummy import DummyOperator
from airflow.models import DAG
from datetime import datetime

# 定义你的DAG1
with DAG(
    dag_id="DAG1",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
    catchup=False
) as dag1:
    # 单个传感器等待DAG2的Task B和Task C
    wait_for_dag2_tasks = ExternalTaskSensor(
        task_id="wait_for_dag2_b_c",
        external_dag_id="DAG2",  # 依赖的外部DAG ID
        external_task_ids=["Task_B", "Task_C"],  # 传入需要等待的任务列表
        allowed_states=["success"],  # 判定任务完成的状态
        failed_states=["failed", "upstream_failed"],  # 如果外部任务失败，传感器也会失败
        poke_interval=30,  # 每隔30秒检查一次任务状态
        mode="reschedule"  # 用reschedule模式节省Worker资源，适合长等待场景
    )

    # 你的Task A（这里用DummyOperator示例）
    task_a = DummyOperator(task_id="Task_A")

    # 设置依赖关系
    wait_for_dag2_tasks >> task_a

几个关键细节要注意 #

• 执行日期对齐：默认情况下，传感器会等待DAG2中与当前DAG1执行日期相同的任务实例。如果你的DAG1和DAG2调度周期一致，这个逻辑完全没问题；如果需要关联不同的执行日期，可以用execution_date_fn参数自定义日期映射逻辑。
• 模式选择：mode="reschedule"会在两次检查之间释放Worker资源，比默认的poke模式更高效，建议等待时间较长时使用。
• 失败处理：如果Task B或Task C进入failed_states中的状态，传感器会直接标记为失败，避免Task A在依赖任务失败的情况下执行。

这个方案既简化了你的DAG结构，也减少了维护成本，完美适配你需要等待多个外部任务的场景。

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