Airflow导出DAG运行统计遇阻：dag_run表end_date字段未填充

我之前也碰到过这个问题！Airflow的dag_run表确实有点坑，end_date字段只有当整个DAG完全成功执行完毕的时候才会被填充——如果DAG还在运行、失败了或者被手动终止，这个字段就会是NULL，这也是你查不到值的原因。下面给你梳理下解决方案：

1. 先搞懂dag_run表的状态逻辑 #

Airflow的dag_run里的state字段直接决定了end_date是否存在：

• success：DAG所有任务都成功，end_date会被设置为最后一个任务完成的时间
• failed：DAG中有任务失败且整个DAG终止，大部分新版本会填充end_date（旧版本可能例外）
• running/queued/scheduled：DAG还在执行或等待调度，end_date为NULL
• up_for_retry/removed：这类状态下end_date同样不会有值

2. 计算DAG运行时长的SQL写法 #

根据你的需求，分两种场景来写查询：

场景1：基于调度时间的快速统计

如果只需要从DAG被调度开始到当前/结束的时长，可以用CASE语句处理不同状态：

SELECT
    run_id,
    dag_id,
    start_date,
    end_date,
    state,
    -- 计算时长（单位：秒）
    CASE
        -- 已完成的DAG（成功/失败）用end_date减start_date
        WHEN state IN ('success', 'failed') THEN EXTRACT(EPOCH FROM (end_date - start_date))
        -- 正在运行的DAG用当前时间减start_date
        WHEN state = 'running' THEN EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - start_date))
        -- 其他状态根据需求处理，比如设为NULL或者0
        ELSE NULL
    END AS total_duration_seconds
FROM dag_run
WHERE dag_id = 'blah';

场景2：基于实际任务执行的精确时长

如果想排除DAG的等待调度时间，只计算实际任务执行的总时长，需要关联task_instance表，取所有任务的最早开始时间和最晚结束时间：

SELECT
    dr.run_id,
    dr.dag_id,
    dr.state,
    -- 实际执行时长：从第一个任务开始到最后一个任务结束的时间差
    EXTRACT(EPOCH FROM (MAX(ti.end_date) - MIN(ti.start_date))) AS actual_execution_seconds,
    -- 同时保留调度时长作为对比
    CASE
        WHEN dr.state IN ('success', 'failed') THEN EXTRACT(EPOCH FROM (dr.end_date - dr.start_date))
        WHEN dr.state = 'running' THEN EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - dr.start_date))
        ELSE NULL
    END AS scheduled_duration_seconds
FROM dag_run dr
INNER JOIN task_instance ti 
    ON dr.dag_id = ti.dag_id 
    AND dr.run_id = ti.run_id
WHERE dr.dag_id = 'blah'
GROUP BY dr.run_id, dr.dag_id, dr.state, dr.start_date, dr.end_date;

3. 额外提醒 #

• 如果你用的是Airflow 1.x版本，dag_run的end_date在DAG失败时可能不会被更新，这时候可以用task_instance的max(end_date)来替代dag_run.end_date
• 导出到InfluxDB时，建议把时长字段转成数值类型（比如秒），方便后续做监控图表

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