如何配置Airflow ExternalSensor检测目标DAG任务后续成功实例触发任务

核心原理 #

原有ExternalSensor默认只会查询和DAG A当前实例execution_date完全匹配的DAG B任务实例，一旦这个固定时间点的B任务实例失败，sensor就会持续等待直到超时。我们只需要打破固定执行时间的匹配限制，允许sensor查询所有晚于目标时间点的B任务实例，只要存在任意成功实例就返回成功即可，全程不需要启用重试机制。

适配方案（Airflow 2.3+ 版本，无需自定义类） #

Airflow 2.3及以上版本的ExternalSensor原生支持多执行日期匹配，仅靠内置参数即可实现需求：

1. 自定义execution_date_fn函数，生成从目标时间到当前查询时间内所有DAG B的调度时间点
2. 开启multiple_dates参数，允许sensor匹配所有生成的时间点对应的B任务实例
3. 只要任意匹配到的实例状态为成功，sensor就会自动通过

示例代码：

from airflow.sensors.external_task import ExternalTaskSensor
from datetime import timedelta
from airflow.utils import timezone

def get_valid_execution_dates(execution_date, **context):
    # 起始时间为原本要匹配的B的10:00实例执行时间
    start_time = execution_date
    # 结束时间为当前sensor的运行时间，覆盖所有后续调度的B实例
    end_time = timezone.utcnow()
    # 按DAG B的调度间隔生成时间点，示例为每小时调度的B的DAG
    valid_dates = []
    current = start_time
    while current <= end_time:
        valid_dates.append(current)
        current += timedelta(hours=1)
    return valid_dates

task_check_b = ExternalTaskSensor(
    task_id="check_b_success_instance",
    external_dag_id="DAG_B",
    external_task_id="your_target_task_id_in_b",
    execution_date_fn=get_valid_execution_dates,
    multiple_dates=True, # 开启多日期匹配
    mode="reschedule", # 非poke模式下不会长期占用worker资源
    poke_interval=300, # 每5分钟查询一次B的任务状态，可按需调整
    timeout=3600 * 24 * 3, # 最长等待3天，可按业务规则调整
    # 不需要配置retries参数，sensor本身会按poke_interval持续查询
)

兼容方案（全版本通用，自定义Sensor类） #

如果使用的是低版本Airflow，没有multiple_dates参数，可以直接继承ExternalSensor重写poke方法实现逻辑：

from airflow.sensors.external_task import ExternalTaskSensor
from airflow.models import TaskInstance
from sqlalchemy import and_

class FlexibleExternalTaskSensor(ExternalTaskSensor):
    def poke(self, context):
        current_exec_date = context["execution_date"]
        # 直接查询所有晚于目标时间的成功B任务实例，存在则返回成功
        success_instance = TaskInstance.query.filter(
            and_(
                TaskInstance.dag_id == self.external_dag_id,
                TaskInstance.task_id == self.external_task_id,
                TaskInstance.execution_date >= current_exec_date,
                TaskInstance.state == "success"
            )
        ).first()
        return success_instance is not None

# 使用方式和原有ExternalSensor一致，不需要额外配置
task_check_b = FlexibleExternalTaskSensor(
    task_id="check_b_success_instance",
    external_dag_id="DAG_B",
    external_task_id="your_target_task_id_in_b",
    mode="reschedule",
    poke_interval=300,
    timeout=3600 * 24 * 3,
)

注意事项 #

• 不要同时配置execution_delta或固定execution_date参数，会和自定义的时间匹配逻辑冲突
• 如果DAG B的调度间隔不固定，优先使用全版本兼容的自定义Sensor方案，不需要提前枚举调度时间点，兼容性更强
• 所有配置均不需要开启重试机制，sensor本身会按设定的poke_interval持续查询直到超时或匹配到成功实例

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