luigi如何处理任务失败？

问题描述

我开始使用luigi。我已经建立了执行多个任务的管道，并且我已经非常小心地确保任务正常运行。因此，管道运行良好。在构建管道时，在任务失败的时候，它们会用:(进行报告，而我对其进行了编辑，直到它们可以正常工作为止。

所以说我有一个做的管道

Task1-->Task2--> Task3

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

现在，假设有5个“ Task1”，5个“ Task2”和一个“ Task3”。所以Task3有点像摘要任务。

我希望每当发生故障时，管道不要停止，而是跳过（并记录故障）并继续下一种情况。（这些“失败”不是因为任务写得不好，而是因为在实际情况下输入到这些任务的数据已损坏）

类似

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

因此，我希望管道执行的是执行任务1和任务2，记录是否存在故障，然后继续并最后用Task3进行总结（甚至包括某种报告存在故障的信息）

如何用luigi做到这一点？

解决方法

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

在路易吉，您有活动。我想您可以使用失败事件并在那里管理错误。您可以生成将用作失败任务的检查点的文件，并记录所需的消息。

官方文档：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
    """
    ...

luigi pipeline pipeline python