如何在Luigi中启用动态需求？

Question

我在路易吉建立了一个任务管道.由于此管道将在不同的上下文中使用,因此可能需要在管道的开头或末尾包含更多任务,或者甚至在任务之间包含完全不同的依赖关系.

就在那时我想:"嘿,为什么要在我的配置文件中声明任务之间的依赖关系？",所以我在config.py中添加了这样的东西:

PIPELINE_DEPENDENCIES = {
     "TaskA": [],
     "TaskB": ["TaskA"],
     "TaskC": ["TaskA"],
     "TaskD": ["TaskB", "TaskC"]
}

我在整个任务中堆积参数时感到很恼火,所以在某些时候我只引入了一个参数,task_config每个参数都存储了Task所需的每个信息或数据run().所以我把它放在PIPELINE_DEPENDENCIES那里.

最后,我将每个Task我定义的继承自两个luigi.Task和一个自定义的Mixin类,它将实现动态requires(),看起来像这样:

class TaskRequirementsFromConfigMixin(object):
    task_config = luigi.DictParameter()

    def requires(self):
        required_tasks = self.task_config["PIPELINE_DEPENDENCIES"]
        requirements = [
            self._get_task_cls_from_str(required_task)(task_config=self.task_config)
            for required_task in required_tasks
        ]
        return requirements

    def _get_task_cls_from_str(self, cls_str):
        ...

不幸的是,这不起作用,因为运行管道给了我以下内容:

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

Scheduled 4 tasks of which:
* 4 were left pending, among these:
    * 4 was not granted run permission by the scheduler:
        - 1 TaskA(...)
        - 1 TaskB(...)
        - 1 TaskC(...)
        - 1 TaskD(...)

Did not run any tasks
This progress looks :| because there were tasks that were not granted run permission by the scheduler

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

还有很多

DEBUG: Not all parameter values are hashable so instance isn't coming from the cache

虽然我不确定这是否相关.

所以:1.我的错误是什么？它可以修复吗？2.还有另一种方法可以达到这个目的吗？

Answer 1

我意识到这是一个老问题，但我最近学习了如何启用动态依赖关系。我能够通过使用 WrapperTask 并生成一个字典理解（如果您愿意的话，您也可以做一个列表）以及我想要传递给 require 方法中其他任务的参数来完成此任务。

像这样的事情：

class WrapperTaskToPopulateParameters(luigi.WrapperTask):
    date = luigi.DateMinuteParameter(interval=30, default=datetime.datetime.today())

    def requires(self):
    base_params = ['string', 'string', 'string', 'string', 'string', 'string']
    modded_params = {modded_param:'mod' + base for base in base_params}
    yield list(SomeTask(param1=key_in_dict_we_created, param2=value_in_dict_we_created) for key_in_dict_we_created,value_in_dict_we_created in modded_params.items())

如果有兴趣，我也可以发布一个使用列表理解的示例。