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.