如何发现在之前的气流DAG任务中是否引发异常？

Question

有没有一种优雅的方法可以检测在同一个dag的前一个任务中是否引发了异常？在

我知道我可以设置TriggerRule =ALL_SUCCESS，因此它只在前面所有任务successfully执行时启动。在

但在我的例子中，我需要触发器规则是ALL_DONE，它将在所有有效状态（包括Failed once）上启动。在

Answer 1

在我看来，您的问题可以通过添加一个中间任务作为第一个任务的下游依赖项来解决。然后将这个中间任务设置为下一个任务的附加upstream_dependency。在

具体如下：

task_1 = Operator()
task_1_on_fail = Operator(trigger_rule='all_failed')
task_1 >> task_1_on_fail

task_2 = Operator()
task_2_on_fail = Operator(trigger_rule='all_failed')
task_2 >> task_2_on_fail

collection_task = Operator(trigger_rule='all_done')
collection_task.set_upstream([task_1, task_1_on_fail, task_2, task_2_on_fail])

在上游任务成功的情况下，_on_fail任务应标记为跳过，这意味着收集任务将正常运行。如果上游任务不成功，_on_fail任务将首先运行，然后收集任务可以继续