如何获取Airflow DAG包含全部任务的整体起始与结束时间

要获取特定DAG Run的整体起止时间（首个任务的实际开始时间、最后一个任务的实际结束时间），可以通过以下几种方式实现：

方法一：通过Airflow UI查看 #

1. 进入Airflow UI，找到目标DAG并点击进入详情页。
2. 在「DAG Runs」列表中，点击目标执行日期对应的条目，进入该DAG Run的详情页面。
3. 页面会展示所有任务实例的状态和时间信息，你可以手动筛选出最早的Start Date和最晚的End Date；也可借助任务时间线视图快速判断整体时间范围。

方法二：使用Airflow CLI + 命令行工具 #

通过airflow tasks list-instances导出任务实例数据，再用jq（需提前安装）处理提取关键时间：

# 替换<your-dag-id>和<your-execution-date>为实际值，执行日期格式如2024-05-20T00:00:00+00:00
airflow tasks list-instances --dag-id <your-dag-id> --execution-date <your-execution-date> --output json | jq '[.[] | select(.start_date != null and .end_date != null)] | min_by(.start_date) as $first | max_by(.end_date) as $last | {"整体开始时间": $first.start_date, "整体结束时间": $last.end_date}'

该命令会自动过滤未完成的任务（无开始/结束时间的），直接输出汇总后的起止时间。

方法三：通过Airflow Python API编写脚本 #

如果需要程序化获取，可编写Python脚本直接查询Airflow元数据：

from airflow.models import DagRun, TaskInstance
from airflow.utils.session import create_session
from datetime import datetime

def get_dag_overall_timestamps(dag_id, execution_date):
    with create_session() as session:
        # 定位目标DAG Run
        dag_run = session.query(DagRun).filter(
            DagRun.dag_id == dag_id,
            DagRun.execution_date == execution_date
        ).first()
        if not dag_run:
            return f"未找到DAG {dag_id}在{execution_date}的运行记录"
        
        # 获取该Run下所有有时间记录的任务实例
        task_instances = session.query(TaskInstance).filter(
            TaskInstance.dag_id == dag_id,
            TaskInstance.execution_date == execution_date,
            TaskInstance.start_date.isnot(None),
            TaskInstance.end_date.isnot(None)
        ).all()
        
        if not task_instances:
            return "该DAG Run下无已完成的任务实例"
        
        # 计算最早开始和最晚结束时间
        earliest_start = min(ti.start_date for ti in task_instances)
        latest_end = max(ti.end_date for ti in task_instances)
        
        return {
            "dag_id": dag_id,
            "execution_date": execution_date.isoformat(),
            "整体开始时间": earliest_start.isoformat(),
            "整体结束时间": latest_end.isoformat()
        }

# 示例调用
if __name__ == "__main__":
    result = get_dag_overall_timestamps("your_dag_id", datetime(2024, 5, 20))
    print(result)

注意：运行脚本时需确保Airflow环境变量（如AIRFLOW_HOME）配置正确。

方法四：直接查询Airflow元数据库 #

如果有权限访问Airflow的元数据库（如PostgreSQL、MySQL），可执行SQL语句直接提取数据：

-- 替换your_dag_id和执行日期为实际值
SELECT
    MIN(start_date) AS dag_overall_start,
    MAX(end_date) AS dag_overall_end
FROM task_instance
WHERE
    dag_id = 'your_dag_id'
    AND execution_date = '2024-05-20 00:00:00'
    AND start_date IS NOT NULL
    AND end_date IS NOT NULL;

关键注意事项 #

• 若存在未完成的任务（end_date为空），上述方法会自动过滤这些任务；若需包含未完成任务的状态，可调整对应过滤条件。
• DagRun表中的start_date和end_date是调度层面的时间，与任务实际执行的起止时间可能存在差异，因此需以TaskInstance表的时间为准。

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