Luigi - 覆盖任务需要/输入

我正在使用 luigi 执行一系列任务，如下所示：

class Task1(luigi.Task):
    stuff = luigi.Parameter()

    def output(self):
        return luigi.LocalTarget('test.json')

    def run(self):
        with self.output().open('w') as f:
            f.write(stuff)


class Task2(luigi.Task):
    stuff = luigi.Parameter()

    def requires(self):
        return Task1(stuff=self.stuff)

    def output(self):
        return luigi.LocalTarget('something-else.json')

    def run(self):
        with self.output().open('w') as f:
            f.write(stuff)

当我像这样启动整个工作流程时，这完全按照预期工作：

luigi.build([Task2(stuff='stuff')])

使用时luigi.build您还可以通过显式传递参数来运行多个任务，按照文档中的这个示例.

但是，在我的情况下，我还希望能够运行以下业务逻辑Task2完全独立于它对工作流程的参与。这对于未实现的任务效果很好requires, 按照这个例子.

我的问题是，如何运行此方法作为工作流程的一部分以及单独运行？显然，我可以添加一个新的私有方法，例如_my_custom_run，它获取数据并返回结果，然后在中使用此方法run，但它只是感觉应该被纳入框架中，所以这让我觉得我误解了 Luigi 的最佳实践（仍在学习框架）。任何建议表示赞赏，谢谢！

听起来像你想要的动态要求。使用该示例中显示的模式，您可以读取配置或传递带有任意数据的参数，并且yield仅根据配置中的字段您想要需要的任务。

# tasks.py
import luigi
import json
import time


class Parameterizer(luigi.Task):
    params = luigi.Parameter() # Arbitrary JSON

    def output(self):
        return luigi.LocalTarget('./config.json')

    def run(self):
        with self.output().open('w') as f:
            json.dump(params, f)

class Task1(luigi.Task):
    stuff = luigi.Parameter()

    def output(self):
        return luigi.LocalTarget('{}'.format(self.stuff[:6]))

    def run(self):
        with self.output().open('w') as f:
            f.write(self.stuff)


class Task2(luigi.Task):
    stuff = luigi.Parameter()
    params = luigi.Parameter()


    def output(self):
        return luigi.LocalTarget('{}'.format(self.stuff[6:]))

    def run(self):

        config = Parameterizer(params=self.params)
        yield config

        with config.output().open() as f:
            parameters = json.load(f)

        if parameters["runTask1"]:
            yield Task1(stuff=self.stuff)
        else:
            pass
        with self.output().open('w') as f:
            f.write(self.stuff)

if __name__ == '__main__':
    cf_json = '{"runTask1": True}'

    print("Trying to run with Task1...")
    luigi.build([Task2(stuff="Task 1Task 2", params='{"runTask1":true}')], local_scheduler=True)

    time.sleep(10)

    cf_json = '{"runTask1": False}'

    print("Trying to run WITHOUT Task1...")
    luigi.build([Task2(stuff="Task 1Did just task 2", params='{"runTask1":false}')], local_scheduler=True)

（这是通过简单地调用来执行的python tasks.py)

我们可以很容易地想象将多个参数映射到多个任务，或者在允许执行各种任务之前应用自定义测试。我们还可以重写它以获取参数luigi.Config.

另请注意以下控制流程Task2:

    if parameters["runTask1"]:
        yield Task1(stuff=self.stuff)
    else:
        pass

在这里，我们可以运行替代任务，或者动态调用任务，正如我们在示例中看到的那样luigi回购。例如：

    if parameters["runTask1"]:
        yield Task1(stuff=self.stuff)
    else:
        # self.stuff is not automatically parsed to int, so this list comp is valid
        data_dependent_deps = [Task1(stuff=x) for x in self.stuff] 
        yield data_dependent_deps

这可能比简单的更复杂一些run_standalone()方法，但我认为这是最接近您在记录的 luigi 模式中寻找的内容。

Source: https://luigi.readthedocs.io/en/stable/tasks.html?highlight=dynamic#dynamic-dependencies