title: How to use an earlier Task's output in Luigi?
body: <p>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.</p>
  
  <p>I've created a simplified example of my setup below.</p>
  
  <pre><code>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()
  </code></pre>
  
  <p>When run via <code>python luigi_example.py CrunchNumbers</code>, 100 random numbers will be created and the list pickled &amp; dumped to disk.  <code>SquareAll</code> loads that pickled list and uses it to require the <code>SquareNumber</code> tasks with the needed parameters.  <code>CubeAll</code> references the same result file for its similar task.</p>
  
  <p>The problem being, when run an exception will be thrown because the <code>numbers.pkl</code> file does not yet exist.  </p>
  
  <p>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.</p>