【问题标题】:handling a lot of parameters in luigi在 luigi 中处理很多参数
【发布时间】:2016-10-12 10:39:48
【问题描述】:

在我的很多项目中,我使用luigi 作为流水线工具。这让我想到用它来实现参数搜索。标准的luigi.file.LocalTarget 有一种非常幼稚的方法来处理参数,这在文档的示例中也有展示:

def output(self):
    return luigi.LocalTarget("data/artist_streams_%s.tsv" % self.date_interval)

即参数保存在文件名中。这使得检查是否已经计算了某个参数组合变得容易。一旦任务的参数变得更复杂,这就会变得混乱。

这是参数搜索的一个非常简单的想法:

import luigi
class Sum(luigi.Task):

    long_ = luigi.Parameter()
    list_ = luigi.Parameter()
    of = luigi.Parameter()
    parameters = luigi.Parameter()

    def output(self):
        return luigi.LocalTarget("task{}_{}_{}_{}.txt".format(self.long_,
                                                              self.list_,
                                                              self.of,
                                                              self.parameters))

    def run(self):

        sum_ = self.long_ + self.list_ + self.of + self.parameters
        with self.output().open('w') as out_file:
            out_file.write(str(sum_))


class ParameterSearch(luigi.Task):

    def requires(self):

        list_of_parameter_combinations = [
            {
                "long_" : 1,
                "list_" : 2,
                "of" : 3,
                "parameters" : 4

            },{
                "long_" : 5,
                "list_" : 6,
                "of" : 7,
                "parameters" : 8
            }
        ]

        for pc in list_of_parameter_combinations:
            yield Sum(**pc)

当然,在这个例子中,所有四个参数都可以编码在文件名中,但不需要很多幻想,这种方法可以达到边界。以类似数组的参数为例。

我的后续想法是将参数和结果存储在某种 信封 对象中,然后可以将其保存为目标。 然后,文件名可以是第一次模糊搜索的参数的某种散列。

有信封类

class Envelope(object):

    @classmethod
    def hashify(cls, params):
        return hash(frozenset(params.items()))

    def __init__(self, result, **params):

        self.params = {}
        for k in params:
            self.params[k] = params.get(k)

    def hash(self):
        return Envelope.hashify(self.params)

然后是新的 Target,它增强了 LocalTarget 并能够检查信封内的所有参数是否匹配:

class EnvelopedTarget(luigi.file.LocalTarget):

    fs = luigi.file.LocalFileSystem()

    def __init__(self, params, path=None, format=None, is_tmp=False):
        self.path = path
        self.params = params

        if format is None:
            format = luigi.file.get_default_format()

        if not path:
            if not is_tmp:
                raise Exception('path or is_tmp must be set')
            path = os.path.join(tempfile.gettempdir(), 'luigi-tmp-%09d' % random.randint(0, 999999999))
        super(EnvelopedTarget, self).__init__(path)
        self.format = format
        self.is_tmp = is_tmp

    def exists(self):
        path = self.path
        if '*' in path or '?' in path or '[' in path or '{' in path:
            logger.warning("Using wildcards in path %s might lead to processing of an incomplete dataset; "
                           "override exists() to suppress the warning.", path)
        if self.fs.exists(path):
            with self.open() as fin:
                envelope = pickle.load(fin)

                try:
                    assert len(envelope.params) == len(self.params)
                    for param,paramval in self.params.items():
                        assert paramval == envelope.params.get(param)

                except(AssertionError):
                    return False
            return True
        else:
            return False

这里的问题是,使用这个目标会增加一些原本 luigi 旨在最小化的样板。我设置了一个新的基础任务

class BaseTask(luigi.Task):

    def output(self, envelope):
        path = '{}{}.txt'.format(type(self).__name__, envelope.hash())
        params = envelope.params
        return EnvelopedTarget(params, path=path)

    def complete(self):

        envelope = Envelope(None, **self.param_kwargs)

        outputs = flatten(self.output(envelope))
        if len(outputs) == 0:
            warnings.warn(
                "Task %r without outputs has no custom complete() method" % self,
                stacklevel=2
            )
            return False

        return all(map(lambda output: output.exists(), outputs))

    def run(self):

        result, outparams = self.my_run()

        envelope = Envelope(result, **outparams)

        with self.output(envelope).open('w') as fout:
            pickle.dump(envelope, fout)

生成的 EnvelopedSum 任务将非常小:

class EnvelopedSum(BaseTask):

    long_ = luigi.Parameter()
    list_ = luigi.Parameter()
    of = luigi.Parameter()
    parameters = luigi.Parameter()

    def my_run(self):
        return sum(self.param_kwargs.values()), self.param_kwargs

此任务可以以与开始时的Sum 任务相同的方式运行。

注意:这个关于如何封装 luigi-task-results 的示例实现远非稳定,而是更多地说明了我所说的封装结果和参数的含义。

我的问题是:难道没有更简单的方法来处理luigi中很多复杂的参数吗?

后续问题:有没有人考虑过保存已执行参数搜索的代码版本(和/或 subtaks 的包版本)的记录?

也感谢任何关于在哪里阅读此主题的 cmets。

注意:

您可能需要一些导入来使其运行:

from luigi.task import flatten
import warnings
import pickle

【问题讨论】:

    标签: python parameter-passing luigi


    【解决方案1】:

    您可能会在邮件列表中获得对此类提案的更好回复。 Luigi 任务代码已经生成了参数的 MD5 哈希值,以生成唯一的任务标识符,您可以获取该标识符。

    luigi/task.py#L128

    # task_id is a concatenation of task family, the first values of the first 3 parameters
    # sorted by parameter name and a md5hash of the family/parameters as a cananocalised json.
    param_str = json.dumps(params, separators=(',', ':'), sort_keys=True)
    param_hash = hashlib.md5(param_str.encode('utf-8')).hexdigest()
    

    【讨论】:

    • 以及如何处理哈希冲突?
    • @bluenote10 应该非常罕见。无论如何,最好将变化最大的参数放在列表的首位,以便在调试时在任务 ID 中提供一些更明显的人类可读差异。
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2020-03-30
    • 2015-06-27
    • 1970-01-01
    • 2017-11-26
    • 2017-06-12
    • 2017-01-22
    • 1970-01-01
    相关资源
    最近更新 更多