有没有一种简单的方法可以在 Airflow 中跳过整个 DAG 运行但仍允许未来运行？答案

【问题标题】：Is there a simple way to skip an entire DAG run in Airflow but still allow future runs?有没有一种简单的方法可以在 Airflow 中跳过整个 DAG 运行但仍允许未来运行？
【发布时间】：2020-06-13 03:07:40
【问题描述】：

我目前有一个每天中午运行的 DAG。我现在希望它在每个星期五和每个月的 10 号运行，以便为自己节省一些成本，但仍然安排它每天运行，以便我可以在任何一天轻松返回并运行 dag。（这背后的原因是有时我需要这个 dag 在特定日期而不是 10 号和每个星期五生成的数据，但我不会提前知道何时需要）。

我的一个想法是使用 BranchPythonOperator 检查是否满足条件，如果满足将触发所有需要启动的任务，否则它会触发一个虚拟任务并跳过其他所有任务。主要问题是这个 DAG 非常庞大并且包含分布在不同文件中的许多任务，所以我正在努力确保一切都依赖于这个 BranchPythonOperator 任务。（关于这个的一个问题，如果我只是让 BranchPythonOperator 独立，不依赖或有任何依赖它，返回任务的名称是否仍然有效？）

我认为必须有一种更简单的方法可以在不满足条件的情况下跳过整个 DAG 运行，而不会阻止我将来在该日期运行。

用示例编辑：

partition_sensor_1 -->
partition_sensor_2 -->
.                         first_dag_task
.
.
partition_sensor_15 -->

因此，在此示例中，我想确保跳过所有这些分区传感器和 DAG 的其余部分。我是否必须将每个分区传感器列为依赖于 Branch python 运算符？这些分区传感器的设置方式通常是在不同的文件中创建的，因此根据 Branching 运算符来设置它们并不容易。希望这是有道理的

【问题讨论】：

标签： python directed-acyclic-graphs airflow-scheduler airflow

【解决方案1】：

我认为你的想法是正确的。但是，您需要在 DAG 的第一个任务中连接 BranchPythonOperator。

所以现在你有：

first_task >> of >> your >> old >> dag >> definition

您首先要定义您的 BranchPythonOperator，然后按如下方式更改 dag：

branch_task >> [dummy_task, first_task]
first_task >> of >> your >> old >> dag >> definition

确保branch_task在条件满足时返回first_task的task_id，否则返回dummy task的task id。

【讨论】：

所以我的问题源于这样一个事实，即有许多“第一个任务”，其中一些是在其他文件中分组创建的，因此很难（也许不可能？）从branch_task 给他们。我不确定是否有必要这样做，如果我只是返回他们的任务名称，即使它们不是依赖项？
仅凭您的文字我很难想象这一点。请您用一个实际的例子来编辑您的问题吗？
所以我有一个 dag，它依赖于大约 15 个其他 dag 来完成，所以有 15 个分区传感器将运行并检查数据是否已降落。我不希望这些在“关闭”的日子里运行，这是否意味着我必须将它们中的每一个都添加为对 Branch 运算符的依赖，或者仅仅不返回它们的任务名称就足够了吗？
当 dag 被调度时，所有的“第一级”任务都会被调度。如果你想让一个 dag 运行，但基本上跳过所有 DAG，我看不到任何其他方法，但实际上有第一级任务来有条件地运行你定义的东西。但是，我可能遗漏了一些东西，值得将您的 DAG 重新考虑为更参数化的方法，例如，手动安排运行并将您希望获取数据的“日期”传递给 DAG。这需要重新设计。否则，解决方案如前所述，将 BranchPythonOperator 作为第一个任务。