Luigi：如何将不同的参数传递给叶子任务？

Question

这是我第二次尝试理解如何在 Luigi 中将参数传递给依赖项。第一个是here。

想法是：我有TaskC，它依赖于TaskB，它依赖于TaskA，它依赖于Task0。我希望整个序列始终完全相同，除了我希望能够控制Task0 读取的文件，我们称之为path。 Luigi 的理念通常是每个任务应该只知道它所依赖的任务及其参数。这样做的问题是TaskC、TaskB 和TaskA 都必须接受变量path，其唯一目的是然后将其传递给Task0。

所以，Luigi 为此提供的解决方案称为Configuration Classes

下面是一些示例代码：

from pathlib import Path
import luigi
from luigi import Task, TaskParameter, IntParameter, LocalTarget, Parameter

class config(luigi.Config):
    path = Parameter(default="defaultpath.txt")

class Task0(Task):
    path = Parameter(default=config.path)
    arg = IntParameter(default=0)
    def run(self):
        print(f"READING FROM {self.path}")
        Path(self.output().path).touch()
    def output(self): return LocalTarget(f"task0{self.arg}.txt")

class TaskA(Task):
    arg = IntParameter(default=0)
    def requires(self): return Task0(arg=self.arg)
    def run(self): Path(self.output().path).touch()
    def output(self): return LocalTarget(f"taskA{self.arg}.txt")

class TaskB(Task):
    arg = IntParameter(default=0)
    def requires(self): return TaskA(arg=self.arg)
    def run(self): Path(self.output().path).touch()
    def output(self): return LocalTarget(f"taskB{self.arg}.txt")

class TaskC(Task):
    arg = IntParameter(default=0)
    def requires(self): return TaskB(arg=self.arg)
    def run(self): Path(self.output().path).touch()
    def output(self): return LocalTarget(f"taskC{self.arg}.txt")

（忽略所有 output 和 run 内容。它们就在那里，因此示例成功运行。）

上面例子的重点是控制print(f"READING FROM {self.path}")这一行，没有任务A、B、C依赖path。

确实，通过配置类，我可以控制Task0 参数。如果Task0 没有传递path 参数，则它采用其默认值，即config().path。

我现在的问题是，在我看来，这似乎只在“构建时”工作，当解释器第一次加载代码时，而不是在运行时（我不清楚细节）。

所以这些都不起作用：

一）

if __name__ == "__main__":
    for i in range(3):
        config.path = f"newpath_{i}"
        luigi.build([TaskC(arg=i)], log_level="INFO")

===== Luigi Execution Summary =====

Scheduled 4 tasks of which:
* 4 ran successfully:
    - 1 Task0(path=defaultpath.txt, arg=2)
    - 1 TaskA(arg=2)
    - 1 TaskB(arg=2)
    - 1 TaskC(arg=2)

This progress looks :) because there were no failed tasks or missing dependencies

===== Luigi Execution Summary =====

我不确定为什么这不起作用。

B)

if __name__ == "__main__":
    for i in range(3):
        luigi.build([TaskC(arg=i), config(path=f"newpath_{i}")], log_level="INFO")

===== Luigi Execution Summary =====

Scheduled 5 tasks of which:
* 5 ran successfully:
    - 1 Task0(path=defaultpath.txt, arg=2)
    - 1 TaskA(arg=2)
    - 1 TaskB(arg=2)
    - 1 TaskC(arg=2)
    - 1 config(path=newpath_2)

This progress looks :) because there were no failed tasks or missing dependencies

===== Luigi Execution Summary =====

这实际上是有道理的。有两个 config 类，我只设法更改了其中一个的 path。

帮助？

编辑：当然，让path 引用全局变量是可行的，但它不是通常 Luigi 意义上的参数。

EDIT2：我尝试了以下答案的第 1) 点：

config 定义相同

class config(luigi.Config):
    path = Parameter(default="defaultpath.txt")

我修正了指出的错误，即Task0 现在是：

class Task0(Task):
    path = Parameter(default=config().path)
    arg = IntParameter(default=0)
    def run(self):
        print(f"READING FROM {self.path}")
        Path(self.output().path).touch()
    def output(self): return LocalTarget(f"task0{self.arg}.txt")

最后我做到了：

if __name__ == "__main__":
    for i in range(3):
        config.path = Parameter(f"file_{i}")
        luigi.build([TaskC(arg=i)], log_level="WARNING")

这不起作用，Task0 仍然得到path="defaultpath.txt"。

Answer 1

因此，您要做的是使用参数创建任务，而不将这些参数传递给父类。这是完全可以理解的，我有时会因为试图处理这个问题而烦恼。

首先，您错误地使用了config 类。使用 Config 类时，如https://luigi.readthedocs.io/en/stable/configuration.html#configuration-classes 中所述，您需要实例化该对象。所以，而不是：

class Task0(Task):
    path = Parameter(default=config.path)
    ...

你会使用：

class Task0(Task):
    path = Parameter(default=config().path)
    ...

虽然现在这可以确保您使用的是值而不是 Parameter 对象，但它仍然不能解决您的问题。在创建类Task0 时，将评估config().path，因此它不会将config().path 的引用分配给path，而是在调用时分配值（始终为defaultpath.txt）。当以正确的方式使用该类时，luigi 将构造一个 Task 对象，其中只有 luigi.Parameter 属性作为新实例的属性名称，如下所示：https://github.com/spotify/luigi/blob/master/luigi/task.py#L436

所以，我看到了两条可能的前进道路。

1.) 第一个是像你一样在运行时设置配置路径，除了将其设置为 Parameter 对象，如下所示：

config.path = luigi.Parameter(f"newpath_{i}")

但是，这需要大量工作才能让您使用 config.path 的任务正常工作，因为现在它们需要以不同的方式接收参数（在创建类时无法评估默认值）。

2.) 更简单的方法是在配置文件中简单地为您的类指定参数。如果你查看https://github.com/spotify/luigi/blob/master/luigi/task.py#L825，你会发现Luigi 中的Config 类实际上只是一个Task 类，所以你可以用它来做任何事情，你可以用一个类来做，反之亦然。因此，你可以在你的配置文件中加入这个：

[Task0]
path = newpath_1
...

3.) 但是，由于您似乎希望使用不同的参数运行多个任务，因此我建议您通过父母传递 args，因为 Luigi 鼓励您这样做。然后你可以运行一切：

luigi.build([TaskC(arg=i) for i in range(3)])

4.) 最后，如果您确实需要摆脱传递的依赖关系，您可以创建一个扩展 luigi.ObjectParameter 的 ParamaterizedTaskParameter，并使用任务实例的 pickle 作为对象。

在上述解决方案中，我强烈建议使用 2 或 3。1 将难以编程，而 4 会创建一些非常难看的参数并且更高级。

编辑：解决方案 1 和 2 更多的是 hack，建议您在 DictParameter 中捆绑参数。