handling a lot of parameters in luigi
<p>In a lot of my projects I use <a href="http://luigi.readthedocs.io/en/stable/" rel="nofollow">luigi</a> as a pipelining tool. This made me think of employing it to implement a parameter search. The standard <code>luigi.file.LocalTarget</code> has a very naive approach to deal with parameters, which is also shown in the doc's examples:</p>

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

<p>Namely, the parameters get saved in the file name. This makes it easy to check if a certain parameter combination is already computed. This gets messy as soon as the parameters for a task are more complex.</p>

<p>Here is the very simple idea of a parameter search:</p>

<pre><code>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)
</code></pre>

<p>Sure, in this example, all four parameters can be encoded in the file name, but it does not require a lot of phantasy, that this approach can reach boundaries. Think for example of array-like parameters.</p>

<p>My follow up idea was to store the parameters and the result in some kind of <strong>envelope</strong> object, which can then be saved as a target.
The file name could then be some kind of hash of the parameters for a first fuzzy search.</p>

<p>There is the envelope class</p>

<pre><code>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)
</code></pre>

<p>Then, there is the new Target, that enhances the LocalTarget and is able to check if all parameters inside the envelope are matching:</p>

<pre><code>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
</code></pre>

<p>The problem here is, that using this target adds some boilerplate which originally luigi aims to minimize. I set up a new base task</p>

<pre><code>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)
</code></pre>

<p>The resulting <code>EnvelopedSum</code> Task would then be pretty small:</p>

<pre><code>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
</code></pre>

<p>This task can be run in the same fashion as the <code>Sum</code> task in the beginning.</p>

<p>Note: this example implementation of how to envelope the luigi-task-results is far from stable and is more of a illustration of what I mean by envelope the results and the parameters.</p>

<p><strong>My question is</strong>: Isn't there a simpler way to deal with a lot of complex parameters in luigi? </p>

<p><strong>Followup-question</strong>: Has anyone thought about holding record of the code-version (and/or package versions of subtaks) on which the parameter search had been performed?</p>

<p>Any comments on where to read on on this topic are also appreciated.</p>

<p><strong>Note:</strong></p>

<p>you probably need some imports to make this running:</p>

<pre><code>from luigi.task import flatten
import warnings
import pickle
</code></pre>

<p>You might get a better response to this sort of proposal on the mailing list. The Luigi task code already does generate an MD5 hash of the parameters to generate a unique task identifier which you could grab.</p>

<p><a href="https://github.com/spotify/luigi/blob/master/luigi/task.py#L79-L82" rel="nofollow">https://github.com/spotify/luigi/blob/master/luigi/task.py#L79-L82</a></p>

<pre><code># 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()
</code></pre>

python parameter-passing luigi