Luigi: Ability to pre-instantiate TaskParameters?

[DanCardin]

Essentially what I'm trying to do is write tasks which can be generic to the tasks before them, so that I can write a LoadToS3 task, for example which doesn't care what target its loading into S3, instead of needing a LoadToS3ForXRequiredTask subclass. So I was hoping to be able to be able to write:

class GenericUpstreamTask(luigi.Task):
    upstream_task = TaskParameter(default=None)
    def requires(self):
        return self.upstream_task

class LoadToS3(GenericUpstreamTask):
    data = ...
    def run(self):
        ...
    def output(self):
        return S3Target(self.input().fn)

class CopyToRedshift(GenericUpstreamTask):
    ...

class ManagerTask(luigi.Task):
    def run(self):       
        results = []
        for dataset in datasets:
            csv_task = GenerateCsv(data=dataset)
            load_to_s3_task = LoadToS3(upstream_task=csv_task)
            copy_to_redshift_task = CopyToRedshift(upstream_task=load_to_s3_task)
            results.append(copy_to_redshift_task)
        yield results

But it doesnt look like TaskParameter works like that. It just want's a Task class. which might work for one level of depth, by adding another parameter to accept any args, and do something like below (except that it won't work nested one level deeper)

    def requires(self):
        self.task_param(**self.task_param_args)

Am I missing something? Or is this sort of workflow atypical?

[Tarrasch]

Can you add a code-example better highlighting how the TaskParameter is used? If your first code-block the definition of a requires()??

[DanCardin]

I dont actually have anything that works in either case. For the first example I updated the original comment to be more of a full example. Upon seeing that not work (because TaskParameter seems to want's the task class instead of an instance), my next attempt was basically

class GenericUpstreamTask(luigi.Task):
    upstream_task = TaskParameter(default=None)
    upstream_task_params = DictParameter(default={})
    def requires(self):
        return self.upstream_task(**self.upstream_task_params)

class LoadToS3(GenericUpstreamTask):
    ...

class CopyToRedshift(GenericUpstreamTask):
    ...

class ManagerTask(luigi.Task):
    def run(self):       
        results = []
        for dataset in datasets:
            results.append(
                CopyToRedshift(
                    upstream_task=LoadToS3,
                    upstream_task_params={
                        upstream_task={
                            upstream_task=GenerateCsv,
                            upstream_task_params={
                                data=dataset,
                            }
                        }
                    }
                )
        yield results

Which is basically the same thing, just deferring the actual instantiation until inside the requires method. And while this would work if I were only trying to compose LoadToS3 and GenerateCsv, this example has a Task inside the upstream_task_params, which is not serializable; so no go here either.

I don't know that I expected either of these methods to work, but I they seem like the most natural way to write tasks which don't have to explicitly define their upstream tasks, in order to be able to be reused among many pipelines, or even among different types of generated files in this pipeline.

Since neither of these methods work (afaict), is there a way to write tasks which are indifferent to their specific input class?

[kalvdans]

I managed to write a Task that takes an instantiated Task as parameter. This is my definition of such parameter:

import luigi
import json

class SubtaskParameter(luigi.Parameter):
    def serialize(self, val):
        return json.dumps((val.get_task_family(), val.param_args))

    def parse(self, val):
        (cls, params) = json.loads(val)
        cls = luigi.task_register.Register.get_task_cls(cls)
        return cls(*params)

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. If closed, you may revisit when your time allows and reopen! Thank you for your contributions.

[SigmaX]

The solution @kalvdans gives is great, but it runs into trouble with parameters that aren't serializable (such as DictParameter).

Personally I'm coming to the conclusion that Luigi just isn't intended to support composable pipelines.

[Tarrasch]

@SigmaX, this is an issue tracker and not a place for rants. If you're willing to cleanly express what you're trying to do there's a discussion Google Group you can use, many other luigi users have gotten fruitful second perspectives over there. Cheers.

[SigmaX]

@Tarrasch Sorry, I didn't mean to slip into opinion.

My understanding is that this topic has already been pretty thoroughly discussed on the Google group (and in the docs, for that matter), so I thought my conclusion might be helpful for other people reading this thread. But you're right, this is an issue tracker, not StackOverflow, and curt remarks don't help anybody anyway, so mea culpa.