任务最后一次成功运行的 Apache Airflow 日期

Question

我正在寻找一种方法来确定上一次成功运行某个任务的时间。我知道有上下文变量 prev_execution_date_success 但根据 docs 这存储了有关整个 DAG 而非组成任务的信息。

我的 DAG 如下所示： DAG 旨在使至少一项任务失败。在失败的情况下，DAG 会重新运行。如果任务已成功运行，我想要实现的是阻止任务实例执行繁重的代码。例如，如果任务 execute_query_heavy_code02 已成功运行（在同一天），则在重新运行 DAG 时，它不应该在第二天运行。

Answer 1

在不更改 DAG 设计的情况下快速修复（不是最佳实践，但足够好）是使用 Airflow Variable，最后成功执行日期为 execute_query_heavy_code02。首先获取变量值并与今天进行比较，如果日期早于今天，则执行繁重的代码并使用set() function设置变量值。

代码将如下所示：

# import section
from airflow.models import Variable
from datetime import datetime

# inside the execute_query_heavy_code02 function
last_execution = Variable.get('heavy_task_last_execution_date')
date_format = '%Y-%m-%d'  # change as you want
last_execution_dt = datetime.strptime(last_execution, date_format).date()
today = datetime.utcnow().date()

if last_execution_dt == today:
    # continue with next task
else:
    # execute heavy code
    Variable.set('heavy_task_last_execution_date', today.strftime(date_format))

如果你想改变设计的某些部分，你可以使用BranchPythonOperator（很好解释here）。