如何获取Airflow中所有DagRun实例的最大/最小/平均运行时长

Airflow原生UI未提供跨所有DAG的DagRun时长聚合统计入口，可通过以下三种方式实现全量实例的最大、最小、平均运行时长统计，覆盖所有DAG关联的全部DagRun范围：

方法1：元数据库直查（临时统计最快方案） #

所有DagRun的生命周期数据都存储在Airflow元库的dag_run表中，通过start_date和end_date的时间差即可计算运行时长。

注意：必须过滤end_date为空的运行中实例，否则会出现时长计算异常

MySQL 版本查询语句 #

SELECT
  MAX(TIMESTAMPDIFF(SECOND, start_date, end_date)) AS max_run_duration_sec,
  MIN(TIMESTAMPDIFF(SECOND, start_date, end_date)) AS min_run_duration_sec,
  ROUND(AVG(TIMESTAMPDIFF(SECOND, start_date, end_date)), 2) AS avg_run_duration_sec
FROM dag_run
WHERE end_date IS NOT NULL
-- 按需开启下一行，仅统计成功实例；注释后统计所有终态实例（含失败、超时）
-- AND state = 'success'
;

PostgreSQL 版本查询语句 #

SELECT
  MAX(EXTRACT(EPOCH FROM (end_date - start_date))) AS max_run_duration_sec,
  MIN(EXTRACT(EPOCH FROM (end_date - start_date))) AS min_run_duration_sec,
  ROUND(AVG(EXTRACT(EPOCH FROM (end_date - start_date))), 2) AS avg_run_duration_sec
FROM dag_run
WHERE end_date IS NOT NULL
-- AND state = 'success'
;

方法2：Airflow 原生Python API 实现（适合脚本/监控集成） #

无需手动配置元库连接，复用Airflow自身的数据库会话即可完成查询，避免账号权限、连接配置的额外操作：

from airflow.models import DagRun
from airflow.utils.session import create_session
from sqlalchemy import func

with create_session() as session:
    query = session.query(
        func.max(func.timestampdiff("SECOND", DagRun.start_date, DagRun.end_date)).label("max_sec"),
        func.min(func.timestampdiff("SECOND", DagRun.start_date, DagRun.end_date)).label("min_sec"),
        func.avg(func.timestampdiff("SECOND", DagRun.start_date, DagRun.end_date)).label("avg_sec")
    ).filter(DagRun.end_date.is_not(None))
    # 按需开启下一行仅统计成功实例
    # query = query.filter(DagRun.state == "success")
    stats = query.first()

print(f"最大运行时长：{stats.max_sec} 秒")
print(f"最小运行时长：{stats.min_sec} 秒")
print(f"平均运行时长：{round(stats.avg_sec, 2)} 秒")

脚本放在Airflow部署节点上，加载Airflow环境变量后直接执行即可。

方法3：自定义CLI命令（适合日常运维高频查询） #

如果需要频繁查询该指标，可以把上述Python逻辑封装为Airflow自定义CLI插件，注册完成后直接执行airflow dagrun-duration-stats即可一键返回结果，无需重复编写脚本。

可选扩展调整 #

• 若需要限定统计时间范围，在过滤条件中增加start_date的时间判断即可，比如仅统计近30天的实例
• 若需要分状态统计时长，在查询中增加GROUP BY state维度，可分别输出成功、失败、超时等不同状态实例的时长指标
• 禁止采用遍历单个DAG再拉取对应DagRun列表的实现方式，全量场景下聚合查询的性能远高于循环遍历

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