气流 DAG 触发器 wait_for_completion 未按预期工作？答案

【问题标题】：Airflow DAG trigger wait_for_completion not working as expected?气流 DAG 触发器 wait_for_completion 未按预期工作？
【发布时间】：2021-06-21 15:44:34
【问题描述】：

我有多个依赖于 initial_dag 的 DAG 首先运行，之后我希望依赖的 DAG 一个接一个地运行。这是我所拥有的：

dag = DAG(
    dag_id=DAG_NAME,
    default_args=default_args,
    schedule_interval=None,
    start_date=airflow.utils.dates.days_ago(1)
)

initial_dag = BashOperator(
    task_id='initial_dag',
    bash_command="python /home/airflow/gcs/dags/task.py",
    dag=dag
)

dependent_dag1 = TriggerDagRunOperator(
    task_id="dependent_dag1",
    trigger_dag_id="dependent_dag1",
    wait_for_completion=True,
    dag=dag
)

dependent_dag2 = TriggerDagRunOperator(
    task_id="dependent_dag2",
    trigger_dag_id="dependent_dag2",
    wait_for_completion=True,
    dag=dag
)

dependent_dag3 = TriggerDagRunOperator(
    task_id="dependent_dag3",
    trigger_dag_id="dependent_dag3",
    wait_for_completion=True,
    dag=dag
)

initial_dag >> dependent_dag1 >> dependent_dag2 >> dependent_dag3

我认为wait_for_completion=True 会在触发下一个 DAG 之前完成每个 DAG 的运行。例如。 initial_dag 运行并完成，然后触发 dependent_dag1 并等待其完成以触发后续任务。

触发 DAG 的顺序是正确的，但它似乎并没有等待前一个 DAG 先完成，例如dependent_dag2 在 dependent_dag1 完成之前被触发。

我错过了什么吗？

【问题讨论】：

标签： airflow airflow-scheduler

【解决方案1】：

这是 Airflow GitHub 讨论板上的 answered，但为了其他用户将两个线程集中在一起。

很遗憾，wait_for_completion 参数在 1.10.x 版本中不可用（请参阅documentation），并且作为通用kwarg 被处理和忽略。此参数从 2.0 开始可用。

【讨论】：

【解决方案2】：

wait_for_completion 参数是任务的完成，而不是 DAG 本身。该任务在成功触发 DAG 时被标记为已完成，因此它不会特别等待该 DAG 完成。

您至少有两个选择：

在触发器调用之间使用ExternalTaskSensor 以等待上一个 DAG 的最后一个任务。
在依赖 DAG 的末尾有一个 TriggerDagRunOperator。例如，dependent_dag1 的最后一个任务将是一个 TriggerDagRunOperator 来运行 dependent_dag2 等等。

您的选择将主要取决于为选项 2 更改 DAG 的可能性，以及您希望拥有的灵活性（认为如果您使用选项 1，您需要跟踪相关 DAG 的最后一个任务，但它是更灵活）。

【讨论】：

【解决方案3】：

如果您需要先等待上一个 DAG 完成，请考虑使用 ExternalTaskSensor 而不是 TriggerDagRunOperator。

在此处查看文档：https://airflow.apache.org/docs/apache-airflow/1.10.3/_api/airflow/sensors/external_task_sensor/index.html

此操作符将等到另一个 DAG（或另一个 DAG 的任务）完成并具有指定状态（默认为“成功”），直到继续前进。

【讨论】：