python - 清除任务气流

Question

我正在使用 Python3 和 Airflow。我有一个日常流程，其中有一些个人任务有时会失败。

如果有人知道，我想知道如何获取失败任务的列表，至少在流程结束时。我可以更改我的代码。我可以以某种方式返回一个包含每个任务名称及其结果的字典，但我认为它必须是一种更简单的方法。

一旦流程“完成”，我希望它清除那些任务状态（我的意思是失败和上游失败的任务）并自动重试它们。

我读过一些关于：

airflow clear -t task_name <dag_name>

所以认为我只需要遍历清除它们的任务名称，但我需要知道它们是谁。

非常感谢！

标签： pythonpython-3.xairflowairflow-scheduler

Answer 1

如果是我，我会编写自己的 Python 脚本，通过加载其模型 ( airflow.models.TaskInstance) 和数据库连接来与 Airflow 交互airflow.settings.Session，定位失败的任务，然后通过脚本清除它们。Airflow 已经有用于清除任务的代码，这些代码可能会提供一个有用的起点。