获取最近一次成功执行 DAG 的日期

Question

我希望在 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 运行的执行。

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

Answer 1

我最终更改了所引用问题中的函数以使用latest_execution_date，这是Airflow中的预定义宏，如下所示：

def get_last_dag_run(dag):
    last_dag_run = dag.latest_execution_date
    if last_dag_run is None: 
        return '2013-01-01'
    else:
        return last_dag_run

目前似乎正在为我工​​作。