Luigi: `output()` executes before `requires()`

Created on 5 Aug 2017 · 6Comments · Source: spotify/luigi

I'm getting an error running the below where output() executes before requires() causing an error because the file doesn't exist. I'm confused why does requires() run before output() and is there a way to deal with the situation?

class BTask(luigi.Task):

    def output(self):
        #==> this executes before requires() with an error
        df_in = pd.read_pickle(self.input()['df_in'].path)
        return [luigi.LocalTarget('file%s.csv'%str(iDate) for iDate in df_in['date'].unique()]

    def requires(self):
        return {'df_in':ATask()}

I've searched the docs Execution Model and Design as well as github but haven't found a similar issue.

Python 3.6.1 |Anaconda custom (64-bit)| (default, May 11 2017, 13:09:58)
[GCC 4.4.7 20120313 (Red Hat 4.4.7-1)] on linux
luigi==2.6.2

Source

citynorman

Most helpful comment

To make sure output() is only ever executed when all dependencies have been completed, you can use the following base class with a custom complete() method, instead of luigi.Task:

from typing import List

from luigi import Target, Task
from luigi.task import flatten


class SafeTask(Task):
    """
    A luigi Task in which `output()` is only ever executed by luigi if all
    dependencies specified in `requires()` have been completed.
    """

    def complete(self) -> bool:
        return all(
            dependency.complete() for dependency in self._dependencies
        ) and all(output.exists() for output in self._outputs)

    @property
    def _dependencies(self) -> List[Task]:
        return flatten(self.requires())

    @property
    def _outputs(self) -> List[Target]:
        return flatten(self.output())

tfiers on 30 Sep 2018

👍3

All 6 comments

output should run before requires(). it runs to check whether the task is already completed

erikbern on 5 Aug 2017

in your example above, you should probably make the df_in a parameter or alternatively use dynamic dependencies on another task where df_in is a parameter

erikbern on 5 Aug 2017

Thanks @erikbern ! I actually call output() in complete() so that makes sense now. Still I wish it would check that requirements are satisfied before calling complete() that's kind of the whole point of luigi to ensure all the dependencies are satisfied.

class BTask(luigi.Task):

    def output(self):
        #==> this executes before requires() with an error
        df_in = pd.read_pickle(self.input()['df_in'].path)
        return [luigi.LocalTarget('file%s.csv'%str(iDate) for iDate in df_in['date'].unique()]

    def requires(self):
        return {'df_in':ATask()}

    def complete(self):
        for iFile in self.output():
            if not os.path.exists(iFile.path):
                return False
        return True

So I fixed it by including the os.path.exists(self.input().path) in complete() at least in my mind it should really be something luigi does for me automatically.

class BTask(luigi.Task):

    def output(self):
        #==> this executes before requires() with an error
        df_in = pd.read_pickle(self.input()['df_in'].path)
        return [luigi.LocalTarget('file%s.csv'%str(iDate) for iDate in df_in['date'].unique()]

    def requires(self):
        return {'df_in':ATask()}

    def complete(self):

        if not os.path.exists(self.input()['df_in'].path):
            return False

        for iFile in self.output():
            if not os.path.exists(iFile.path):
                return False
        return True

citynorman on 5 Aug 2017

👍1

i think this seems a bit un-idiomatic. off the top of my head what I would do is something like

let BTask() depend on ATask()
inside BTask.run, read the df_in of ATask(), and require a number of third task CTask(iDate)
do all the processing inside the CTask

erikbern on 5 Aug 2017

To make sure output() is only ever executed when all dependencies have been completed, you can use the following base class with a custom complete() method, instead of luigi.Task:

from typing import List

from luigi import Target, Task
from luigi.task import flatten


class SafeTask(Task):
    """
    A luigi Task in which `output()` is only ever executed by luigi if all
    dependencies specified in `requires()` have been completed.
    """

    def complete(self) -> bool:
        return all(
            dependency.complete() for dependency in self._dependencies
        ) and all(output.exists() for output in self._outputs)

    @property
    def _dependencies(self) -> List[Task]:
        return flatten(self.requires())

    @property
    def _outputs(self) -> List[Target]:
        return flatten(self.output())

tfiers on 30 Sep 2018

👍3

Is it the only use of output() to be called from complete() to check if the task is completed?