如何在Airflow中获取DAG启动时间、结束时间及运行时长？

方法1：直接使用Airflow UI内置功能 #

Airflow 2.x及以上版本的DAG Run列表（进入对应DAG后点击"DAG Runs"标签）已经自带Duration列，直接显示每个DAG Run的运行时长，同时也会展示Start Date和End Date，完全不需要额外开发，这是最简便的方式。

方法2：通过Operator上下文获取并推送到XCom #

如果需要自定义计算或在UI的任务详情中查看，可以通过Airflow的上下文获取DagRun对象的内置属性，无需访问数据库：

代码示例 #

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

def compute_dag_runtime(**kwargs):
    # 从上下文获取当前DAG Run实例
    dag_run = kwargs["dag_run"]
    start_dt = dag_run.start_date
    end_dt = dag_run.end_date

    if end_dt:
        runtime = end_dt - start_dt
        # 将时长推送到XCom，可在UI的任务详情页查看
        kwargs["ti"].xcom_push(key="dag_total_runtime", value=str(runtime))
        print(f"DAG总运行时长: {runtime}")
    else:
        print("当前DAG Run仍在执行中")

with DAG(
    dag_id="track_dag_runtime",
    start_date=datetime(2024, 5, 1),
    schedule_interval="@daily",
    catchup=False,
    tags=["runtime", "utility"]
) as dag:
    track_runtime_task = PythonOperator(
        task_id="compute_runtime",
        python_callable=compute_dag_runtime,
        # 开启上下文传递
        provide_context=True,
    )

查看方式 #

执行完任务后，进入该任务的详情页，切换到XCom标签，就能看到dag_total_runtime对应的时长值。

方法3：使用模板宏直接在Operator中打印 #

如果只需要在任务日志中查看时长，可以利用Airflow的模板宏，无需编写Python函数：

BashOperator示例 #

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    dag_id="print_dag_runtime",
    start_date=datetime(2024, 5, 1),
    schedule_interval="@daily",
    catchup=False
) as dag:
    print_runtime_task = BashOperator(
        task_id="print_runtime",
        bash_command='''
            echo "DAG启动时间: {{ dag_run.start_date }}"
            echo "DAG结束时间: {{ dag_run.end_date if dag_run.end_date else '执行中' }}"
            echo "运行时长: {{ dag_run.end_date - dag_run.start_date if dag_run.end_date else '计算中' }}"
        '''
    )

查看方式 #

任务执行后，进入任务详情页的Logs标签，就能看到打印的时长信息。

这些方法都无需直接操作Airflow数据库，完全利用Airflow内置的对象和功能，实现简单且符合Airflow的原生逻辑。

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