子类中的 Luigi 子依赖项

Question

我有一个流程 A->B->C，其中 C 依赖于 B，B 依赖于 A
假设我有另一个流程 D->B->C

我尝试重用该任务。如何轻松重用？

我可以创建一个继承任务 B 的子类并将需求更改为任务 D，但是为了允许 D->B->C，我需要再次创建继承任务 C 的子类并将需求方法更改为 B 的子类.

这很麻烦，我想知道有没有简单的方法可以做到这一点> 或者这是 luigi 的正确行为？

Answer 1

这没有意义，可能是太抽象了。 B 或 C 如何知道它是否需要（）任务 A 或任务 D？ A 和 D 是否产生相同类型的输出？如果有不同的情况，您应该以改变任务要求的方式对它们进行编码。根据您提供的非常有限的信息，这是一个部分的、过于简化的示例。

class TaskC(luigi.Task):
    tasktype = luigi.Parameter() 
    def requires():
        return TaskB(tasktype=self.tasktype)

class TaskB(luigi.Task):
    tasktype = luigi.Parameter()
    def requires():
       if self.tasktype == "A":
          return TaskA()
       else:
          return TaskD()
    def run():
       file = self.input().open('r')
       #do something with output from either