Airflow DAG触发时间控制：非阻塞延迟执行问题排查

你的问题核心在于TimeSensor的目标时间没有基于DAG触发时的固定时间计算，而是每次检查时动态生成变化的时间，导致陷入无限等待。以下是修正后的实现，采用非阻塞的TimeSensor模式，同时确保目标时间固定：

关键思路 #

1. 先通过PythonOperator获取DAG触发时的UTC时间（dag_run.start_date），而非传感器执行时的当前时间
2. 根据触发时间判断是否需要延迟：若在UTC 16:00-16:30区间，目标时间设为当天16:30；否则设为触发时间（立即执行）
3. 将目标时间存入XCom，供TimeSensor使用，确保目标时间固定不变
4. TimeSensor使用mode='reschedule'模式，等待期间释放Worker资源，避免阻塞

完整代码示例 #

from airflow import DAG
from airflow.sensors.time_sensor import TimeSensor
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import pendulum

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1, tzinfo=pendulum.UTC),
    'retries': 0,
}

def calculate_target_time(**context):
    # 获取DAG触发的UTC时间
    trigger_time = context['dag_run'].start_date.astimezone(pendulum.UTC)
    trigger_date = trigger_time.date()
    trigger_hour = trigger_time.hour
    trigger_minute = trigger_time.minute
    
    # 判断是否在延迟区间内
    if trigger_hour == 16 and 0 <= trigger_minute < 30:
        # 设置目标时间为当天UTC 16:30
        target_time = datetime.combine(trigger_date, datetime.min.time(), tzinfo=pendulum.UTC) + timedelta(hours=16, minutes=30)
    else:
        # 立即执行，目标时间为触发时间
        target_time = trigger_time
    
    # 将目标时间存入XCom
    context['ti'].xcom_push(key='target_time', value=target_time.isoformat())

def get_target_time(**context):
    # 从XCom取出预先计算好的目标时间
    target_time_str = context['ti'].xcom_pull(task_ids='calculate_target_time', key='target_time')
    return pendulum.parse(target_time_str)

with DAG(
    'dag_A',
    default_args=default_args,
    schedule_interval=None,  # 由外部DAG触发，无需定时
    catchup=False,
    tags=['triggered_dag'],
) as dag:
    # 第一步：计算目标执行时间
    calc_target = PythonOperator(
        task_id='calculate_target_time',
        python_callable=calculate_target_time,
        provide_context=True,
    )

    # 第二步：非阻塞等待到目标时间
    wait_for_time = TimeSensor(
        task_id='wait_until_target_time',
        target_time_callable=get_target_time,
        poke_interval=60,  # 每分钟检查一次时间是否到达
        mode='reschedule',  # 释放Worker资源的非阻塞模式
    )

    # 第三步：执行你的核心业务逻辑
    def execute_main_logic(**context):
        print("开始执行dag-A的核心任务...")
        # 替换为你的实际业务代码

    main_task = PythonOperator(
        task_id='main_business_task',
        python_callable=execute_main_logic,
        provide_context=True,
    )

    # 任务依赖关系
    calc_target >> wait_for_time >> main_task

为什么之前的代码会无限循环？ #

如果你的TimeSensor直接在target_time_callable中使用当前时间（比如datetime.utcnow()）计算延迟，会导致每次传感器检查时，目标时间都会被重新设置为未来的某个时间，永远无法到达，从而陷入无限等待。而上述方案通过预先计算固定的目标时间，彻底避免了这个问题。

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