python - 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 将构造一个只有 luigi.Parameter 属性的 Task 对象作为新实例的属性名称，如下所示: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.) 但是，由于您似乎想要运行多个任务并为每个任务设置不同的参数，我建议您像 Luigi 鼓励的那样通过父级传递参数。然后你可以运行一切:

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

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

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

编辑:解决方案 1 和 2 比任何东西都更像是 hack，建议您将参数捆绑在 DictParameter 中。