Airflow问题：DAG运行未按预期执行，如何设置时间偏移？

这是预期行为吗？ #

是的，这是Airflow的预期调度逻辑。

Airflow的调度核心是基于调度周期的边界时间触发任务，而非从start_date开始累加间隔：

• 当你设置schedule_interval = '*/5 * * * *'时，cron表达式定义的周期边界是0、5、10、15...分钟点，任务会在这些边界时间到达后触发，执行的是上一个5分钟周期的任务（比如00:05触发的是00:00-00:05周期的任务）。
• start_date仅用于控制第一个任务的可调度时间下限，不会修改cron表达式本身的触发时间点。你的start_date是00:01，第一个满足“周期结束时间≥start_date”的边界点是00:05，因此任务从00:05开始触发，后续按0、5、10...的间隔运行。

如何设置时间偏移？ #

有三种常用方式实现自定义时间偏移：

1. 直接修改cron表达式（最简单） #

如果要偏移1分钟，将cron的分钟位改为1-59/5，这样会直接指定在1、6、11、16...分钟点触发任务：

schedule_interval = '1-59/5 * * * *'

如果需要其他偏移量（比如3分钟），只需调整起始数字：3-59/5 * * * *，对应3、8、13...分钟触发。

2. 使用timedelta作为调度间隔（适合非cron的固定间隔） #

当使用timedelta替代cron表达式时，Airflow会从start_date开始累加间隔来计算触发时间：

from datetime import timedelta

start_date = pendulum.datetime(2022, 1, 1, 0, 1, tz="UTC")
schedule_interval = timedelta(minutes=5)

此时任务的触发时间会是00:01、00:06、00:11...（注意：如果开启了catchup=True，Airflow会补跑start_date到当前时间之间的所有间隔任务；若不需要补跑，需设置catchup=False）。

3. 使用自定义Timetable（Airflow 2.2+，灵活度最高） #

Airflow 2.2及以上版本支持自定义Timetable，比如用DeltaTimetable实现基于start_date的偏移调度：

from airflow.timetables.delta import DeltaTimetable
from datetime import timedelta

start_date = pendulum.datetime(2022, 1, 1, 0, 1, tz="UTC")
schedule = DeltaTimetable(delta=timedelta(minutes=5), start_date=start_date)

这种方式完全基于start_date累加间隔，不受cron边界限制，适合复杂的偏移需求。

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