Задачи Луиджи, входящие в бесконечные циклы - PullRequest
Новая
       18

Задачи Луиджи, входящие в бесконечные циклы

0 голосов
/ 15 марта 2019

У меня есть простая задача 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
...
...
...

Что мне здесь не хватает?

Спасибо, Ойшик

1 Ответ

1 голос
/ 19 марта 2019

Я бы посоветовал вам взглянуть на документацию по методу 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)
...