How to use an earlier Task's output in Luigi?

Viewed 1166

I'm writing a pipeline in which later tasks need to read the output of earlier tasks, so they can know what parameters they need to pass in their requirements.

I've created a simplified example of my setup below.

import random
import pickle
import luigi


class WriteNumbers(luigi.Task):

    def requires(self):
        pass

    def run(self):
        # pickle a random list of ints 1-100
        numbers = [random.randint(1, 100) for _ in range(100)]
        pickle.dump(numbers, open("./numbers.pkl", 'wb'))

    def output(self):
        return luigi.LocalTarget("./numbers.pkl")


class SquareNumber(luigi.Task):
    number = luigi.IntParameter()

    def requires(self):
        pass

    def run(self):
        # given a number as the parameter, write a file containing its square
        with open("./squared_{}".format(self.number), 'w') as f:
            f.write(str(self.number ** 2))

    def output(self):
        return luigi.LocalTarget("./squared_{}".format(self.number))


class SquareAll(luigi.WrapperTask):

    def requires(self):
        yield WriteNumbers()  # require the number list to be pickled first
        numbers = pickle.load("./numbers.pkl")  # load the number list
        for n in numbers:  # square each number in the number list
            yield SquareNumber(number=n)

class CubeNumber(luigi.Task):
    number = luigi.IntParameter()

    def requires(self):
        pass

    def run(self):
        # given a number as the parameter, write a file containing its cube
        with open("./cubed_{}".format(self.number), 'w') as f:
            f.write(str(self.number ** 3))

    def output(self):
        return luigi.LocalTarget("./cubed_{}".format(self.number))


class CubeAll(luigi.WrapperTask):

    def requires(self):
        yield WriteNumbers()  # require the number list to be pickled first
        numbers = pickle.load("./numbers.pkl")  # load the number list
        for n in numbers:  # square each number in the number list
            yield CubeNumber(number=n)

class CrunchNumbers(luigi.WrapperTask):
    def requires(self):
        yield SquareAll()
        yield CubeAll()

if __name__ == '__main__':
    luigi.run()

When run via python luigi_example.py CrunchNumbers, 100 random numbers will be created and the list pickled & dumped to disk. SquareAll loads that pickled list and uses it to require the SquareNumber tasks with the needed parameters. CubeAll references the same result file for its similar task.

The problem being, when run an exception will be thrown because the numbers.pkl file does not yet exist.

How can I allow later tasks to generate dependencies based on the output of an earlier task? I have used random numbers here to indicate that the output cannot be known ahead of time: my actual application is crunching data from an API.

1 Answers

You are using Dynamic Dependencies and these need to be called from the run method (when the results of the requires are available as input), so CubeAll and SquareAll should be structured like this:

class SquareAll(luigi.WrapperTask):

    def requires(self):
        yield WriteNumbers()  # require the number list to be pickled first

    def run(self):
        numbers_file = self.input()[0].path
        numbers = pickle.load(numbers_file)  # load the number list
        for n in numbers:  # square each number in the number list
            yield SquareNumber(number=n)
Related