我如何在 Python Luigi 任务管理器中表示子流？

Question

假设我有流程 A->B。此流程的用户需要 B。假设我有一个任务 C，它需要 B，但我希望 D 在流程 A->B 之前发生。我如何在 Luigi 中执行此操作？

Answer 1

一种解决方案是将 C 的要求传递到子流 A->B:

import luigi

class A(luigi.Task):
  precedes = luigi.Parameter()

  def requires(self):
     return self.precedes

class B(luigi.Task):
  precedes = luigi.Parameter()

  def requires(self):
     return A(precedes = self.precedes)

class D(luigi.Task):
  pass

class C(luigi.Task):
  def requires(self):
      return B(precedes = D())