Airflow - 如何从上次成功运行的实例中执行 DAG？

如何解决Airflow - 如何从上次成功运行的实例中执行 DAG？

我有一个每小时运行一次的 DAG 并执行如下增量加载：

select * from <table_name>
where last_mod_dt between <execution_date> AND <next_execution_date>;

---***execution_date 这里是当前dag实例的执行日期（1PM,2PM...）

然而，一些每小时 DAG 的实例在两者之间失败，因此要确保下一个任务实例执行总是只从最后一个成功的实例中选择；这样就不会从源中丢失任何增量记录。

例如：- 当前场景

DAG_Ins1 - 1 PM - Success (last_mod_dt between 1 PM AND 2 PM)
DAG_Ins2 - 2 PM - Fail (last_mod_dt between 2 PM AND 3 PM)
DAG_Ins3 - 3 PM - Fail (last_mod_dt between 3 PM AND 4 PM)
DAG_Ins4 - 4 PM - Sucesss (last_mod_dt between 4 PM AND 5 PM)

最后一个成功的第四个实例只选择当前的执行日期（不管最后的状态如何）

预期：

DAG_Ins1 - 1 PM - Success (last_mod_dt between 1 PM AND 2 PM)
DAG_Ins2 - 2 PM - Fail (last_mod_dt between 2 PM AND 3 PM)
DAG_Ins3 - 3 PM - Fail (last_mod_dt between 2 PM AND 4 PM)
DAG_Ins4 - 4 PM - Sucesss (last_mod_dt between 2 PM AND 5 PM)

但是，最后的第 4 个实例应该从上次成功执行结束状态（在上例中为下午 2 点）中选择日期。

不想查询 Airflow 元数据表。执行此方法：- get_latest_execution_date() 返回 DAG 的最后一次成功/失败的日期时间？任何其他 {{macro}} 可以做必要的事情吗？

谢谢！！

解决方法

我认为您正在寻找的是 prev_execution_date_success 宏。 此宏提供上次成功运行 DAG 的 execution_date。

您的 SQL 可以是：

select * from <table_name>
where last_mod_dt between '{{ prev_execution_date_success }}' AND '{{ next_execution_date }}';

可以在 here 中找到所有可用宏的参考。