python - 任务失败后 Airflow 停止调度 dagruns

Question

如果 dagrun 出现故障，我对 Airflow 的作用感到困惑。我想要实现的行为是:

1. DAG 的定期触发(每小时)
2. 重试任务
3. 如果任务失败 n 次重试，发送有关失败的电子邮件
4. 当下一个每小时触发时，触发一个新的 dagrun，就好像什么都没有失败一样。

这些是我的 dag 参数和任务参数:

任务默认值:

'depends_on_past': True,
'start_date': airflow.utils.dates.days_ago(2),
'email': ['email@address.co.uk'],
'email_on_failure': True,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
'wait_for_downstream': False,

dag 参数:

schedule_interval=timedelta(minutes=60),
catchup=False,
max_active_runs=1

我想我误解了其中的一些论点，因为在我看来，如果一个任务失败 n 次(即 dagrun 失败)，那么下一个 dagrun 会被安排但只是永远处于运行状态并且永远不会再有 dagrun成功(甚至被安排)。例如，这里是 dagrun(我不知道在哪里可以找到基于文本的调度程序日志，如 this question)，其中 dag 计划每 5 分钟而不是每小时运行一次:

每5分钟执行一次，直到失败，之后最后一次执行才处于运行状态，过去30分钟一直如此。

我做错了什么？

我应该补充一点，重新启动调度程序没有帮助，手动将正在运行的任务设置为失败也没有帮助...

Answer 1

您将 depends_on_past 设置为 True，这会阻止启动下一个 DagRun。

From the docs :depends_on_past (bool) – 当设置为 true 时，任务实例将按顺序运行，同时依赖于前一个任务的计划来成功。 start_date 的任务实例允许运行。

这意味着您的 Dag 正在尝试运行，但它正在等待，直到来自先前 DagRun 的相应任务具有成功状态。