在 Python 的 Airflow 中，如何在特定时间后停止任务运行？

Question

我正在尝试使用 Python 的 Airflow 库。我希望它定期抓取网页。

我遇到的问题是，如果我的start_date 是几天前的，那么当我启动调度程序时，它将从start_date 回填到今天。例如：

假设今天是本月的 20 号。

假设start_date 是本月的 15 号。

如果我在 20 号启动调度程序，它会在 20 号刮掉页面 5 次。它将看到一个 DAG 实例应该在 15 日运行，并将在 20 日运行该 DAG 实例（15 日的实例）。然后它将在 20 日运行 16 日的 DAG 实例，以此类推。

简而言之，Airflow 会尝试“赶上”，但这对于网络抓取没有意义。

有什么方法可以让 Airflow 在一定时间后认为 DAG 实例失败？

Answer 1

此功能在 Airflow 的路线图中，但目前不存在。

见： Issue #1155

您可以使用BranchPythonOperator 拼凑出一个解决方案。正如文档中所说，确保您已设置depends_on_past=False（这是默认设置）。我没有设置气流，所以目前无法测试并为您提供示例代码。

Answer 2

Airflow 的设计考虑到了“回填”，因此路线图项违背了其主要逻辑。

现在您可以为这个特定任务或整个 dag 更新 start_date。

每个运营商都有一个 start_date http://pythonhosted.org/airflow/code.html#baseoperator

调度程序不是为停止而设计的。如果您今天运行它，您可以将您的任务 start_date 设置为今天，这对我来说似乎是逻辑。