如何获取 Airflow 作业的最后两个成功执行日期？

Question

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

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

Answer 1

您可以利用 SQLAlchemy 魔法来检索 execution_dates 以对抗最后 'n' 次成功运行

from pendulum import Pendulum
from typing import List, Dict, Any, Optional
from airflow.utils.state import State
from airflow.settings import Session
from airflow.models.taskinstance import TaskInstance

def last_execution_date(
    dag_id: str, task_id: str, n: int, session: Optional[Session] = None
) -> List[Pendulum]:
    """
    This function is to queries against airflow table and
    return the most recent execution date
    Args:
        dag_id: dag name
        task_id : task name
        n : number of runs
        session: Session to connect airflow postgres db
    Returns:
        list of execution date of most of recent n runs
    """
    query_val = (
        session.query(TaskInstance)
        .filter(
            TaskInstance.dag_id == dag_id,
            TaskInstance.task_id == task_id,
            TaskInstance.state == State.SUCCESS,
        )
        .order_by(TaskInstance.execution_date.desc())
        .limit(n)
    )
    execution_dates: List[Pendulum] = list(map(lambda ti: ti.execution_date, query_val))
    return execution_dates

# Above function can be used as utility function and can be leveraged with provide_session as below:
 
last_success_date_fn = provide_session(last_execution_date) # can use provide session decorator as is.

这个 sn-p 是端到端测试的，可以在 prod 中使用。

我参考了views.py 中的tree() method 来提出这个脚本。

---

或者，您可以将此 SQL 查询触发到 Airflow 的元数据库，以检索成功运行的最后 n 个执行日期

SELECT execution_date
FROM task_instance
WHERE dag_id = 'my_dag_id'
  AND task_id = 'my_task_id'
  AND state = 'success'
ORDER BY execution_date DESC
LIMIT n