如何识别 Airflow 中所有正在运行的 dag 实例中最后完成的 dag答案

【问题标题】：How to identify the last completed dag among all the running dag instances in Airflow如何识别 Airflow 中所有正在运行的 dag 实例中最后完成的 dag
【发布时间】：2021-07-15 07:36:23
【问题描述】：

它是一个基于调度的 DAG。气流创建了 4 个 dag 实例，我想确定最后一个完成的 dag 实例，即。 d-instance-2。在所有 dag-instance 结束时完成。

气流中是否有任何方法可以识别所有正在运行的 dag 实例中最后完成的 dag？

dag-instance	dag - run order	dag-completion order
d-instance - 1	run - 1	complete - 1 - first (need to identify)
d-instance - 2	run - 2	complete - 4 - last (need to identify)
d-instance - 3	run - 3	complete - 2
d-instance - 4	run - 4	complete - 3

dag_args = {
"depends_on_past": False,
"start_date": datetime(2017, 1, 1),
"end_date": datetime(2020, 12, 31)}


dag = DAG("Multiple_dag_instance will trigger",
      description='dag_description',
      default_args=dag_args,
      schedule_interval="@yearly")

chain(
  task_1,
  task_2,
  identify_the_final_dag_completion_task)

在这里，start_date: datetime(2017, 1, 1) 和 end_date: datetime(2020, 12, 31) 创建 2017、2018、2019 和 2020 年的 4 个年度 dag 实例。

【问题讨论】：

您能解释一下 DAG 的上下文和使用吗？为什么需要识别最后运行的实例？为什么要将task_2 与identify_the_final_dag_completion_task 链接起来？这些是什么类型的运算符？
可以通过 UI 中的 SQL 查询访问此信息。是否足以满足您的需求？
@bruno-uy，想在所有 dags 实例成功完成结束后触发通知任务。它基于PythonOperator

标签： python airflow airflow-scheduler

【解决方案1】：

您可以使用python代码查询与DAG id对应的Dag Runs并按执行开始日期过滤，然后按结束日期对dag进行排序并选择您想要的。

您可以尝试类似以下的方法：

from airflow.models.dagrun import DagRun
from datetime import datetime

# Your execution date here. Maybe ds injected with Jinja templates.
exec_date = datetime.strptime('2021-05-19', '%Y-%m-%d')

dag_ids = ['dag_id_1', 'dag_id_2']

runs = DagRun.find(dag_ids, execution_start_date=exec_date)
runs.sort(key=lambda x: x.end_date)

我在运行气流 2.1.0 的主机上的 python shell 上对此进行了测试并且工作正常。我相信它应该在操作员的执行中对您同样有效。

【讨论】：

它为我们提供了所有正在运行的dags信息，要求是识别dag的所有实例都成功完成，然后我有一个触发/回调链内的通知任务。
@eigenharsha 这很奇怪。它应该向您显示所有 DagRuns，无论是否运行。您甚至应该能够按其状态进行过滤——运行、成功等API Docs。你用对了吗execution_start_date？
@eigenharsha 还有：你如何等待 DAG？是这些 subdags 还是你真的触发了一个新的 dag？我认为最后一个任务应该以某种方式等待所有 dag 完成然后查询它们，否则如果您启动并继续，它们肯定会运行。