如何获得在气流中运行的 dag 的最新执行时间 [英] how to get latest execution time of a dag run in airflow

问题描述

我尝试了下面的代码，但仍然出现问题

I tried below code but still i am getting issue

from airflow.models DagModel

def get_latest_execution_date(**kwargs):

session = airflow.settings.Session()

f = open("/home/Insurance/InsuranceDagsTimestamp.txt","w+")

try:
    Insurance_last_dag_run = session.query(DagModel)
    for Insdgrun in Insurance_last_dag_run:
        if Insdgrun is None: 
            f.write(Insdgrun.dag_id+",9999-12-31"+"\n")
        else:
            f.write(Insdgrun.dag_id+","+ Insdgrun.execution_date+"\n")
except:
    session.rollback()
finally:
    session.close()

t1 = PythonOperator(
    task_id='records',
    provide_context=True,
    python_callable=get_latest_execution_date,
    dag=dag)

有什么办法可以修复和获取最新的 dag 运行时信息

Is there any way how to fix and get the latest dag run time information

推荐答案

有多种方法可以获取 DagRun 的最新执行情况.一种方法是利用 Airflow DagRun 模型.

There are multiple ways to get the most recent execution of a DagRun. One way is to make use of the Airflow DagRun model.

from airflow.models import DagRun

def get_most_recent_dag_run(dag_id):
    dag_runs = DagRun.find(dag_id=dag_id)
    dag_runs.sort(key=lambda x: x.execution_date, reverse=True)
    return dag_runs[0] if dag_runs else None


dag_run = get_most_recent_dag_run('fake-dag-id-001')
if dag_run:
    print(f'The most recent DagRun was executed at: {dag_run.execution_date}')

您可以在 位于此处的气流文档.

这篇关于如何获得在气流中运行的 dag 的最新执行时间的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！