TaskClassAmbigiousException in Luigi when yielding tasks in run()

user_6165007 0
Viewed 449

I'm struggling with an error in Luigi that I don't understand. I don't know if this is a known issue, a limitation of Luigi or if I am doing something wrong.

I am using Luigi in a real problem with many tasks and many dependences. However, I have made a toy example in which this issue appears clearly.

Let us consider two Tasks, TaskA and TaskB, TaskA requiring the execution of two previous instances of TaskB with differents values of a Luigi Parameter.

If I code the dependences in the requires() method of TaksA, then nothing bad happens. All three tasks execute and I get my exit files written.

But if I code the dependences in the run() method of TaksA, then I get the ugly TaskClassAmbigiousException.

In my real problem, I cannot yield the task in the requires() method, because I need to know the result of a previous task that is also yielded in the requieres() method, so I tried to yield the task in the run() and got the same exception.

Ok, here is the code of the toy example. First, yielding the task in requieres(), and it works:

import luigi

class TaskB(luigi.Task):
    j = luigi.IntParameter(default=1)

    def output(self):
        return luigi.LocalTarget("data/outputB{j}.txt".format(j=self.j))

    def requires(self):
        pass

    def run(self):
        print_file = 'TaskB' + str(self.j)

        with self.output().open('w') as out_file:
            out_file.write(print_file)

class TaskA(luigi.Task):
    i = luigi.IntParameter(default=1)

    def output(self):
        return luigi.LocalTarget("data/outputA{i}.txt".format(i=self.i))

    def requires(self):
        yield TaskB(j=self.i)
        yield TaskB(j=self.i+1)

    def run(self):
        print_file = ""
        for input_target in self.input():
            with input_target.open('r') as in_file:
                for line in in_file:
                    print_file+=line + 'TaskA' + str(self.i)

        with self.output().open('w') as out_file:
            out_file.write(print_file)
               
if __name__ == '__main__':
   taskA = TaskA(i=2)

And second, yielding the task in run(), and I get this:

 File "/home/ppo0011l/.conda/envs/nudge/lib/python3.6/site-packages/luigi/worker.py", line 1081, in _handle_next_task
    for module, name, params in new_requirements]

  File "/home/ppo0011l/.conda/envs/nudge/lib/python3.6/site-packages/luigi/worker.py", line 1081, in <listcomp>
    for module, name, params in new_requirements]

  File "/home/ppo0011l/.conda/envs/nudge/lib/python3.6/site-packages/luigi/task_register.py", line 251, in load_task
    task_cls = Register.get_task_cls(task_name)

  File "/home/ppo0011l/.conda/envs/nudge/lib/python3.6/site-packages/luigi/task_register.py", line 181, in get_task_cls
    raise TaskClassAmbigiousException('Task %r is ambiguous' % name)

The code:

import luigi

class TaskB(luigi.Task):
    j = luigi.IntParameter(default=1)

    def output(self):
        return luigi.LocalTarget("data/outputB{j}.txt".format(j=self.j))

    def requires(self):
        pass

    def run(self):
        print_file = 'TaskB' + str(self.j)

        with self.output().open('w') as out_file:
            out_file.write(print_file)

class TaskA(luigi.Task):
    i = luigi.IntParameter(default=1)

    def output(self):
        return luigi.LocalTarget("data/outputA{i}.txt".format(i=self.i))

    def requires(self):
        pass

    def run(self):
        print_file = ""
        target1 = yield TaskB(j=self.i)
        target2 = yield TaskB(j=self.i+1)
        for input_target in [target1, target2]:
            with input_target.open('r') as in_file:
                for line in in_file:
                    print_file+=line + 'TaskA' + str(self.i)

        with self.output().open('w') as out_file:
            out_file.write(print_file)
               
if __name__ == '__main__':
   taskA = TaskA(i=2)
   luigi.build([taskA], workers=1,local_scheduler=True,log_level='WARNING')

EDIT: I edit to add another related question. As what I want to do is to yield a task with a parameter that depends on an earlier yielded task, if this were possible for me it will be enough:

  def requires(self):
        taskb_target = yield TaskB(j=self.i)
        taskb_target.open('r')
# do something and yield next Task depending on what taskb_target has
        yield TaskB(j=self.i+1)

But unforntunately this does not work. Luigi says "NoneType" object has no attribute "open".

However, when you yield a task in the run() method, you can access the output in runtime. It seems that there is a big asymmetry...

SECOND EDIT:

I have been doing more trials and I have found a weird conclusion: the second piece of code that I wrote in the original question (when in a .py file) can be executed ad eternum, even if deleting the output files and so forcing luigi to reexecute the tasks. However, the first piece of code can be excecuted only once (adn then, on the first execution, it works!!). But if you delete the files and execute the code again, you will get the ambigous task error.

I think it has something to do with the Register object of luigi. But what is really confusing for me is that this behaviour is different whether I yield the taskB in the requieres or the run method.

What I still don't know if is the problem arises when redefining class Tasks that are already in the Register module of luigi. It could be that... I tried also to place the class definitions in a .py different from the main .py, but when running twice it breaks. The only way to run properly is restarting the kernel, and you have only one chance!

luigi python
2 Answers

When you use yield, you do not get a return value, because you are basically returning a value from a co-routine. I'm actually surprised that using yield inside of requires works for you, as it has caused things to crash for me. What you want to do is first define and then yield the task. So, for instance, you would have:

class TaskA(luigi.Task):
    def run(self):
        task_1 = TaskB(j=self.i)
        yield task_1
        with task_1.output().open('r') as in_file:
            # Get data

        task_2 = TaskB(j=self.i+1, ...)
        yield task_2
        ...
user_11377130

Ok, now I found that the issue arises only when executing twice the script. And doing trial an error, I found that the problem was that, when importing TaskA and TaskB again, they are registred again in luigi.task_register.Register.

In fact, there is an attribute of Register, _reg, that contains all registered classes. And in the second excecution of the module, TaskA and TaskB are registered again. I don't know why. It's weird but true. And it only happens when importing TaskA, which is even weirder.

So a way to fix this problem that I found is the following:

import luigi
from luigi import task_register

class TaskB(luigi.Task):
    j = luigi.IntParameter(default=1)

    def output(self):
        return luigi.LocalTarget("data/outputB{j}.txt".format(j=self.j))

    def requires(self):
        pass

    def run(self):
        print_file = 'TaskB' + str(self.j)

        with self.output().open('w') as out_file:
            out_file.write(print_file)

class TaskA(luigi.Task):
    i = luigi.IntParameter(default=1)

    def output(self):
        return luigi.LocalTarget("data/outputA{i}.txt".format(i=self.i))

    def requires(self):
        pass

    def run(self):
        print_file = ""
        target1 = yield TaskB(j=self.i)
        target2 = yield TaskB(j=self.i+1)
        for input_target in [target1, target2]:
            with input_target.open('r') as in_file:
                for line in in_file:
                    print_file+=line + 'TaskA' + str(self.i)

        with self.output().open('w') as out_file:
            out_file.write(print_file)

taskA_list = [c for c in task_register.Register._reg if c.__name__ == 'TaskA']
if len(taskA_list) > 1:
   task_register.Register._reg.pop()
   task_register.Register._reg.pop()
               
if __name__ == '__main__':
   taskA = TaskA(i=2)
   luigi.build([taskA], workers=1,local_scheduler=True,log_level='WARNING')

This is a trick that works and make the module reexecutable forever either deleting or not the output files. But obviously is not the most elegant solution. I write it just to help luigi developers to fix it, or if I am doing something wrong to correct me!

user_61650070
Related