如何解决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
"""
...
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。