4. DAG调度与时间语义

在Airflow中，时间语义是理解调度行为、编写幂等任务、执行补数回填的核心基础。很多初学者会误以为DAG在某个时间点运行，处理的就是“当前时间”的数据，但Airflow的设计哲学恰恰相反：调度周期结束时才触发运行，而运行处理的是刚刚结束的那个完整周期内的数据。理解这一“错位”关系，是掌握Airflow时间语义的关键。

一、核心原理：调度周期与逻辑时间

Airflow中每个DAG Run都绑定一个逻辑时间，它代表该次运行所对应的数据窗口的起点。这个逻辑时间在Airflow 2.2之前称为execution_date，之后引入logical_date作为更准确的别名，两者在大多数场景下指向同一个值。与之配套的还有data_interval_start和data_interval_end，它们共同描述了一个左闭右开的时间区间[start, end)。

关键规则是：DAG Run的触发时刻 = data_interval_end。也就是说，当调度周期走到终点时，Airflow才创建这次运行，并把logical_date设为data_interval_start。例如一个每天0点调度的DAG，在1月2日0点触发的那次运行，其logical_date是1月1日0点，data_interval_start是1月1日0点，data_interval_end是1月2日0点，处理的是1月1日全天的数据。

二、方法流程：四个时间参数的关系与获取方式

在任务函数中，可以通过context字典或模板变量获取这些时间。下面是一个典型的PythonOperator示例：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

def process(context):
    logical = context["logical_date"]
    start = context["data_interval_start"]
    end = context["data_interval_end"]
    print(f"logical_date={logical}")
    print(f"data_interval_start={start}")
    print(f"data_interval_end={end}")

with DAG(
    dag_id="time_semantics_demo",
    start_date=datetime(2024, 1, 1),
    schedule="0 0 * * *",
    catchup=False,
) as dag:
    PythonOperator(
        task_id="show_time",
        python_callable=process,
    )

在模板中也可以直接使用：{{ logical_date }}、{{ data_interval_start }}、{{ data_interval_end }}。对于schedule=None的手动触发DAG，data_interval_start和data_interval_end可能相等或为None，需要做兼容处理。

下表总结了不同调度周期下四个参数的取值规律：

调度周期	触发时刻	logical_date	data_interval_start	data_interval_end
每天0点	1月2日00:00	1月1日00:00	1月1日00:00	1月2日00:00
每小时	10:00	09:00	09:00	10:00
每周一0点	1月8日00:00	1月1日00:00	1月1日00:00	1月8日00:00
每月1日0点	2月1日00:00	1月1日00:00	1月1日00:00	2月1日00:00

可以看到，logical_date始终等于data_interval_start，而data_interval_end等于下一次调度周期的起点，也就是本次运行的触发时刻。

三、案例场景：用时间参数实现幂等增量处理

假设有一个每天0点调度的DAG，需要从业务库抽取前一天的数据写入数仓。正确做法是使用data_interval_start和data_interval_end作为过滤条件，而不是使用datetime.now()。

def extract(context):
    start = context["data_interval_start"]
    end = context["data_interval_end"]
    sql = f"""
        SELECT * FROM orders
        WHERE created_at >= '{start}'
          AND created_at <  '{end}'
    """
    # 执行抽取并写入分区 dt={start.date()}

这样写的好处是：无论任务在何时被重跑、补数或手动触发，只要logical_date不变，抽取的数据窗口就完全一致，天然具备幂等性。补数时只需指定logical_date，Airflow会自动推导出对应的data_interval_start和data_interval_end。

四、误区与复盘

误区一：把execution_date当作实际执行时间。这是最常见的误解。execution_date（即logical_date）是数据窗口的起点，不是任务真正运行的墙上时间。任务可能在逻辑时间之后几小时甚至几天才被调度执行。

误区二：在任务中使用datetime.now()做数据过滤。这会导致重跑时数据窗口漂移，破坏幂等性。正确做法始终是使用data_interval_start和data_interval_end。

误区三：认为data_interval_start和data_interval_end总是相差一个调度周期。对于schedule=None或@once的DAG，这两个值可能相等或为None，代码中需要做防御性判断。

误区四：混淆catchup与时间语义。catchup=True时，Airflow会从start_date开始，按调度周期逐个创建DAG Run，每个Run的logical_date依次递增。理解这一点，才能正确控制补数范围。

掌握logical_date、data_interval_start、data_interval_end与调度周期的关系，是从“能跑通DAG”迈向“能写出可靠、可补数、幂等的数据管道”的关键一步。在实际项目中，建议统一使用data_interval_start和data_interval_end作为数据过滤边界，把logical_date作为分区键和补数入口，从而构建出清晰、可维护的时间语义体系。