如何忽略在另一个任务的 run() 中触发的 Luigi 任务失败

Posted 2023-03-31

【中文标题】如何忽略在另一个任务的 run() 中触发的 Luigi 任务失败

【英文标题】：How to ignore failures on Luigi tasks triggered inside another task's run()

【发布时间】：2019-04-30 14:10:37

【问题描述】：

考虑以下任务：

import luigi


class YieldFailTaskInBatches(luigi.Task):
    def run(self):
        for i in range(5):
            yield [
                FailTask(i, j)
                for j in range(2)
            ]


class YieldAllFailTasksAtOnce(luigi.Task):
    def run(self):
        yield [
            FailTask(i, j)
            for j in range(2)
            for i in range(5)
        ]

class FailTask(luigi.Task):
    i = luigi.IntParameter()
    j = luigi.IntParameter()

    def run(self):
        print("i: %d, j: %d" % (self.i, self.j))
        if self.j > 0:
            raise Exception("i: %d, j: %d" % (self.i, self.j))

如果j > 0，FailTask 将失败。 YieldFailTaskInBatches 在 for 循环中多次生成 FailTask，而 YieldAllFailTasksAtOnce 生成数组中的所有任务。

如果我运行YieldFailTaskInBatches，Luigi 会运行在第一个循环中产生的任务，并且当其中一个失败 (i = 0, j = 1) 时，Luigi 不会产生其余的任务。如果我运行YieldAllFailTasksAtOnce，Luigi 会按预期运行所有任务。

我的问题是：我怎样才能告诉 Luigi 继续运行 YieldFailTasksInBatches 上的剩余任务，即使某些任务失败了？有可能吗？

我问的原因是我有大约 40 万个任务要触发。我不想一次触发所有任务，因为这会让 Luigi 花费太多时间来构建每个任务的要求（他们可以有 1 到 400 个需求）。我目前的解决方案是批量生产它们，一次很少，但是如果其中任何一个失败，任务就会停止，其余的不会产生。

看来this issue如果实施的话可以解决这个问题，但我想知道是否还有其他方法。

【问题讨论】：

它与***.com/questions/48150406/… 有关，但他们的解决方法对我不起作用。

【参考方案1】：

这很hackish，但它应该做你想做的事：

class YieldAll(luigi.Task):
    def run(self):
        errors = list()
        for i in range(5):
            for j in range(2):
                try:
                    FailTask(i, j).run()
                except Exception as e:
                    errors.append(e)

        if errors:
            raise ValueError(f' all traceback: errors')

class FailTask(luigi.Task):
    i = luigi.IntParameter()
    j = luigi.IntParameter()

    def run(self):
        print("i: %d, j: %d" % (self.i, self.j))
        if self.j > 0:
            raise Exception("i: %d, j: %d" % (self.i, self.j))

所以基本上你是在 luigi 上下文之外运行任务。除非你输出一个目标，否则 luigi 永远不会知道任务是否已经运行。

luigi 唯一知道的任务是 YieldAll。如果任何 YieldAll 产生错误，代码将捕获它并将 YieldAll 任务设置为失败状态。

【讨论】：

不会让requires 中的任务导致它们在执行之前都被安排好？这是我要避免的，因为我有 40 万多个任务。

@VitorBaptista 这很 hackish 但有可能，调整答案

谢谢！我同意它超级hacky，我可能不会在生产中使用它，但你的帮助非常有用。我想这是做我想做的事情的唯一方法，除非github.com/spotify/luigi/issues/2516 被实施......

如果FailTask 也有requires 方法，你知道我能做什么吗？如果我在第 7 行使用yield FailTask，则异常传播，使用FailTask.run() 不会执行requires 部分。