A simple multiworkers pipeline in Luigi

multiple_workers.py

import luigi

NUMBER_OF_FILES = 50


class WritePipelineTask(luigi.Task):
    flow_id = luigi.Parameter()

    def output(self):
        return luigi.LocalTarget("data/output_one_{}.txt".format(self.flow_id))

    def run(self):
        with self.output().open("w") as output_file:
            output_file.write("pipeline")


class AddMyTask(luigi.Task):
    flow_id = luigi.Parameter()

    def output(self):
        return luigi.LocalTarget("data/output_two_{}.txt".format(self.flow_id))

    def requires(self):
        return WritePipelineTask(flow_id=self.flow_id)

    def run(self):
        with self.input().open("r") as input_file:
            line = input_file.read()

        with self.output().open("w") as output_file:
            decorated_line = "My "+line
            output_file.write(decorated_line)


class AllTasks(luigi.WrapperTask):

    def requires(self):
        for i in range(NUMBER_OF_FILES):
            yield AddMyTask(flow_id=str(i))