luigi 如何处理任务失败？答案

【问题标题】：how does luigi handles task failure?luigi 如何处理任务失败？
【发布时间】：2020-10-29 10:28:55
【问题描述】：

我开始使用 Luigi。我已经建立了一个可以执行多项任务的管道，并且我已经足够小心以确保任务正常运行。所以管道运行良好。在构建管道时，在任务失败的时候，他们被:( 报告，我编辑它们直到它们工作正常。

假设我有一个管道

Task1-->Task2--> Task3

在这种情况下，如果 Task2 失败，则不会执行 Task3，并且管道会停止。通常是因为写入Task2时出错。

现在假设有 5 个“Task1”、5 个“Task2”和一个“Task3”。所以 Task3 是一种总结性的任务。

我希望我的管道在出现故障时不要停止，而是跳过（并可能记录故障）并继续下一个案例。（这些“失败”不是因为任务写得不好，而是因为假设输入到这些任务的数据在真实案例中被破坏了）

类似

你可以看到Task1在Task2之前执行。标记为红色的任务是“失败”。

所以我希望管道做的是执行任务 1 和任务 2，记录是否有故障并继续，最后用 Task3 进行总结（甚至包括某种报告有故障）

我怎样才能用 Luigi 做到这一点？

【问题讨论】：

标签： python pipeline luigi

【解决方案1】：

我认为你必须在你的可选任务中添加一个 try/except 语句，这样当它们失败时它们仍然会生成一个虚拟输出并且 luigi 认为它们很好。

【讨论】：

【解决方案2】：

在路易吉你有事件。我想您可以使用失败事件并管理其中的错误。您可以生成将用作失败任务的检查点的文件并记录您喜欢的消息。

官方文档：https://luigi.readthedocs.io/en/stable/tasks.html#events-and-callbacks

@luigi.contrib.hadoop.JobTask.event_handler(luigi.Event.FAILURE)
def mourn_failure(task, exception):
    """Will be called directly after a failed execution
       of `run` on any JobTask subclass
    """
    ...

【讨论】：