如何防止气流回填 dag 运行？

Question

假设你有一个气流 DAG，它没有回填的意义，这意味着，在它运行一次之后，运行随后的快速运行将完全没有意义。

例如，如果您要从某些仅每小时更新一次的源中将数据加载到数据库中，则快速连续发生的回填只会一次又一次地导入相同的数据。

当你实例化一个新的每小时任务时，这尤其烦人，它在开始之前运行s N 它错过的每个小时的次数，做冗余工作运行在您指定的时间间隔内。

我能想到的唯一解决方案是他们在 FAQ of the docs

中特别建议反对的事情

We recommend against using dynamic values as start_date, especially datetime.now() as it can be quite confusing.

有什么方法可以禁用 DAG 的回填，或者我应该执行上述操作吗？

Answer 1

这似乎是一个未解决的气流问题。我知道我真的很想拥有完全相同的功能。据我所知，这是；它可能对其他人有用。

UI 功能（至少在 1.7.1.3 中）可以帮助解决这个问题。如果您转到树视图并单击特定任务（方框），将出现一个带有 'mark success' 按钮的对话框按钮。单击 'past'，然后单击 'mark success' 会将 DAG 中该任务的所有实例标记为成功，而不是运行。顶级 DAG（顶部的圆圈）也可以以类似的方式标记为成功，但似乎没有办法标记多个 DAG 实例。

我还没有深入研究它，但可以使用 'trigger_dag' 子命令来标记 DAG 的状态。看这里：https://github.com/apache/incubator-airflow/pull/644/commits/4d30d4d79f1a18b071b585500474248e5f46d67d

用于标记 DAG 的 CLI 功能正在开发中： http://mail-archives.apache.org/mod_mbox/airflow-commits/201606.mbox/%3CJIRA.12973462.1464369259000.37918.1465189859133@Atlassian.JIRA%3E https://github.com/apache/incubator-airflow/pull/1590

更新（2016 年 9 月 28 日）：添加了一个新的运算符 'LatestOnlyOperator' (https://github.com/apache/incubator-airflow/pull/1752)，它只会运行最新版本的下游任务。听起来很有用，希望它能尽快发布

更新 2：自 airflow 1.8 起，LatestOnlyOperator 已发布。

Answer 2

升级到 airflow 版本 1.8 并在 airflow.cfg 中使用 catchup_by_default=False 或将 catchup=False 应用于每个 dag。

https://github.com/apache/incubator-airflow/blob/master/UPDATING.md#catchup_by_default

Answer 3

在你的 dag 声明中设置 catchup=False 将提供这个确切的功能。

我没有 "reputation" 发表评论，但我想说 catchup=False 正是（由我）为此目的设计的。此外，我可以验证在 1.10.1 中它在实例化中明确设置时是否正常工作。但是，当放置在默认参数中时，我看不到它的工作。不过，我已经离开 Airflow 18 个月了，所以我还需要一段时间才能了解为什么默认参数不能用于追赶。

dag = DAG('example_dag',
        max_active_runs=3,
        catchup=False,
        schedule_interval=timedelta(minutes=5),
        default_args=default_args)

如何防止气流回填 dag 运行？

How to prevent airflow from backfilling dag runs?

python

scheduled-tasks

airflow