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

Question

我有一个每小时运行一次的 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)

最后 4 个成功的实例只选择当前执行日期（与最后状态无关）

预期：

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}} 可以满足需要吗？

谢谢！！

Answer 1

我认为您正在寻找的是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找到。

Airflow >= 2.2.0 更新：

prev_execution_date_success 宏已弃用。您可以使用其他宏，例如：

select * from <table_name>
where last_mod_dt between '{{ prev_data_interval_end_success }}' AND '{{ data_interval_start }}';

注意到 Airflow 2.2.0 实现了AIP 39 - Richer schedule_interval，因此直接查询数据间隔可能更有意义：

select * from <table_name>
where last_mod_dt between '{{ data_interval_start }}' AND '{{ data_interval_end }}';