Airflow 中实现“定时执行 + 前置任务成功依赖”的最佳实践-Python教程-PHP中文网

Airflow 中实现“定时执行 + 前置任务成功依赖”的最佳实践

在 Airflow 中，可通过组合定时调度、任务依赖与 TimeSensorAsync（或 TimeDeltaSensorAsync）实现：让 T08:30 任务仅在当日 T05:30 任务成功完成且到达指定时间后才触发，兼顾时序控制与状态依赖。

在 airflow 中，可通过组合定时调度、任务依赖与 timesensorasync（或 timedeltasensorasync）实现：让 t08:30 任务仅在当日 t05:30 任务成功完成 *且* 到达指定时间后才触发，兼顾时序控制与状态依赖。

Airflow 的 DAG 级调度（schedule_interval 或 schedule）决定了整个工作流的触发节奏，而任务级依赖（>>）确保执行顺序与状态约束。要同时满足「每日固定时间触发」和「仅当同日前置任务成功才继续」两个条件，不能仅靠 DAG 调度或单纯上下游依赖——因为默认调度下，T08:30 任务若独立设为每天 08:30 触发，将无视 T05:30 的当日执行结果。

✅ 正确解法是：将整个 DAG 设为每日 05:30 触发（即以 T05:30 为起点），并在其后插入一个时间传感器，使 T08:30 任务延迟至当天 08:30 才运行，且仅在 T05:30 成功的前提下推进。

以下是推荐实现（基于 Airflow 2.6+，使用 TimeSensorAsync 提升效率）：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.sensors.time import TimeSensorAsync  # 异步版，不阻塞 Worker
from datetime import datetime, time

default_args = {
    "owner": "airflow",
    "retries": 1,
}

with DAG(
    dag_id="t0530_then_t0830",
    default_args=default_args,
    schedule="0 5 * * *",  # 每日 05:30 UTC 触发整个 DAG（注意：Airflow 默认用 UTC）
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["timing", "dependency"],
) as dag:

    def task_t0530():
        print("Running T05:30 job — e.g., data ingestion")
        return "success"

    t0530 = PythonOperator(
        task_id="T05:30",
        python_callable=task_t0530,
    )

    # 等待到当日 08:30（UTC），但仅在 t0530 成功后才开始等待
    wait_until_0830 = TimeSensorAsync(
        task_id="wait_until_0830",
        target_time=time(8, 30),  # 注意：target_time 是本地时间（即 DAG timezone 下的时间）
        mode="reschedule",  # 推荐：释放 worker slot，避免长时间占用
    )

    def task_t0830():
        print("Running T08:30 job — e.g., downstream analytics")
        return "completed"

    t0830 = PythonOperator(
        task_id="T08:30",
        python_callable=task_t0830,
    )

    # 关键依赖链：T05:30 → 等待 → T08:30
    t0530 >> wait_until_0830 >> t0830

? 关键说明与注意事项：

✅ 时区对齐：TimeSensorAsync.target_time 解析为 DAG 所配置的 timezone（如未显式设置，默认为 UTC）。若需按北京时间（CST/UTC+8）执行，请在 DAG 中声明：
```
from airflow.utils.timezone import utc
from pendulum import timezone
tz_beijing = timezone("Asia/Shanghai")
# 然后在 DAG 参数中加入：timezone=tz_beijing
```
⚠️ 避免 TimeSensor（同步版）：它会持续轮询并占用一个 Worker Slot，尤其在长等待（如 3 小时）时资源浪费严重；TimeSensorAsync 基于异步事件驱动，更高效可靠。
? catchup=False 必须启用：防止历史调度堆积导致多个“当日”实例并发，破坏「仅执行一次」语义。
? 替代方案（高级场景）：若需跨多日依赖（如“T08:30 必须等 T05:30 在 同一逻辑日期 成功”，而 DAG 运行时间因延迟偏移），可结合 ExternalTaskSensor + 自定义 execution_date_fn，但本例中单一 DAG + 时间传感器已足够简洁健壮。