Luigi 任务进入无限循环答案

【问题标题】：Luigi Tasks going into infinite LoopsLuigi 任务进入无限循环
【发布时间】：2019-03-15 16:09:25
【问题描述】：

我有一个简单的 luigi 任务，它在运行时会产生一个不同的参数，如下所示。

import luigi

class ComputeJob(luigi.Task):

   id_parameter = luigi.parameter.IntParameter()

    #run defination
    def run(self):


        print ("\nrunning task {}".format(self.id_parameter))
        #Do some work here

        if self.id_parameter < 10: 
            next_val = self.id_parameter + 1
            yield ComputeJob(id_parameter = next_val)

我希望它运行 10 次然后退出运行，但在执行第 10 次迭代后，它从一开始就开始重新执行 9 个步骤。因此，任务在第 9 步和第 10 步中不断循环。

所以预期的输出应该是：

running task 1
running task 2
running task 3
running task 4
running task 5
running task 6
running task 7
running task 8
running task 9
running task 10

但我得到的输出是：

running task 1
running task 2
running task 3
running task 4
running task 5
running task 6
running task 7
running task 8
running task 9
running task 10
running task 9
running task 10
running task 9
running task 10
...
...
...

我在这里错过了什么？

谢谢，Oyshik

【问题讨论】：

所以，我发现每次调用 Task 时，都会在输出目标上调用 exists()。如果目标存在，则不会重新运行该任务。我希望我可以一起取消输出，因为我的任务不会产生任何输出。

标签： python-3.x luigi

【解决方案1】：

我建议您查看方法 complete 的文档，如果您想“一起取消输出”，如您在后续评论中所述。

另一种选择是创建一个额外的包装器任务来运行这个您已经创建的 ComputeJob 任务，您可以根据需要进行多次。

import luigi

class ComputeJob(luigi.Task):
  id_parameter = luigi.parameter.IntParameter()
  done = False

  #run definition
  def run(self):
    print ("\nrunning task {}".format(self.id_parameter))
    #Do some work here
    self.done = True

  def complete(self):
    if self.done:
      return True
    else:
      return False

class RunComputeJobs(luigi.WrapperTask):
  def requires(self):
    for i in range(1,10):
      yield ComputeJob(id_parameter = i)

【讨论】：