主页 / coding / 问题 / 77222079
Accepted
rholdberh
Asked: 2023-10-03 19:59:59 +0800 CST

检查是否有两个或多个任务失败

  • 772

我正在尝试建立这样的逻辑：

task_1 >> state
task_2 >> state
task_3 >> state

在“状态”中，我需要检查，如果三分之二的任务失败，那么 DAG 应该失败。此外，仅当所有“tasks_x”完成时才应执行状态。

我怎样才能做到这一点？

airflow

1 个回答

  1. Best Answer

    您可以使用TaskInstance获取失败任务的列表。这是一个例子：

    def state(ti: TaskInstance, **kwargs):
    # find failed tasks
    failed = ti.get_dagrun().get_task_instances(state=TaskInstanceState.FAILED)

    if len(failed) >= 2:
        raise Exception('State failed')
    # processing...
    • 0

相关问题