Apache Airflow如何获取两次DAG run执行时间差避免重复处理文件

1. 可查询的核心时间字段说明 #

Airflow的DAG Run实例自带3个通用时间字段，可根据场景选择使用：

• execution_date：DAG Run的逻辑执行日期，属于调度唯一标识，同一调度周期的execution_date全局唯一，不会受运行重试、延迟启动影响
• start_date：DAG Run实际启动运行的时间，会受资源排队、重试逻辑影响
• end_date：DAG Run运行完成的时间，仅成功/失败的终态实例会有该值

如果是为了避免重复处理文件做幂等校验，优先选择execution_date或者成功实例的end_date，过滤掉运行失败、正在运行的无效实例。

2. 具体实现方案 #

方案1：DAG代码内直接查询（适配业务逻辑校验场景） #

直接在DAG的Python代码中调用Airflow内置ORM接口查询历史运行记录，计算得到时间差后可直接用于文件处理的前置校验：

1. 导入依赖

from airflow.models import DagRun
from datetime import timedelta

2. 封装时间差计算函数

def calc_dag_run_time_diff(dag_id: str, **context) -> timedelta:
    # 按执行日期倒序查询当前DAG所有成功运行的历史实例，取最近1条
    last_success_run = DagRun.find(
        dag_id=dag_id,
        state="success",
        order_by=DagRun.execution_date.desc()
    )
    # 首次运行无历史记录，返回超大时间差默认值，直接走正常处理逻辑
    if not last_success_run:
        return timedelta(days=365)
    # 取上一次成功运行的逻辑时间，和当前运行的逻辑时间做差
    last_run_exec_time = last_success_run[0].execution_date
    current_run_exec_time = context["execution_date"]
    return current_run_exec_time - last_run_exec_time

3. 业务逻辑前置校验示例
   在处理文件的Operator前添加判断，如果两次运行时间差小于文件的最小生成周期，直接跳过处理逻辑即可避免重复消费。

方案2：CLI命令查询（适配人工排查场景） #

如果是离线排查重复处理问题，可直接在Airflow部署节点执行以下命令查询最近两次成功运行的时间，手动计算差值：

airflow dags list-runs -d 替换为你的DAG_ID --state success --limit 2

返回结果会展示两次实例的execution_date、start_date、end_date，直接相减即可得到时间差。

3. 幂等校验补充建议 #

如果要进一步降低重复处理的概率，可搭配双重校验逻辑：

• 第一重用时间差做粗过滤，过滤掉间隔小于文件生成周期的运行实例
• 第二重记录已处理文件的文件名+修改时间戳到元数据库，每次处理前先校验是否存在已处理记录

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