相关疑难解决方法(0)

我需要获取 Airflow 作业的最后两个成功执行日期以在当前运行中使用。示例：执行日期 作业状态 2020-05-03 成功 2020-05-04 失败 2020-05-05 成功

问题：当我在 5 月 6 日运行作业时，我应该将 5 月 3 日和 5 日的值放入变量中。是否可以？

我们有以下命令，我们可以清除失败的任务并一次尝试重新运行它们

airflow clear [-s START_DATE] [-e END_DATE] --only_failed dag_id

有什么方法可以从所有 dags 中获取所有失败任务的信息并将其导出到文件（excel 或文本）中

我希望在 Airflow 中创建一个转换，并且我想确保从上次运行 DAG 以更新我的目标表以来从我的源中获取所有数据。为此，我希望能够获得最近成功的执行。

我发现了这一点：Apache 气流宏获得最后一次 dag 运行执行时间，这使我到达最终目标的某个地方，但是，这只会获得 DAG 执行的最后一次，无论它是否成功。

SELECT col1, col2, col3
FROM schema.table
WHERE table.updated_at > '{{ last_dag_run_execution_date(dag) }}';

如果执行失败（由于连接或类似原因），last_dag_run_execution_date(dag) 将更新，但我们错过了之前 DAG 运行的执行。

理想情况下，这将拉取最近的非失败执行。或者如果有人有任何想法我怎么能满足这个，请让我知道

我试过下面的代码，但我仍然遇到问题

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 运行时信息