multinode_luigi_workflow_test.py

import luigi
import time

class SleepA(luigi.Task):
    taskid = luigi.Parameter()

    def output(self):
        return luigi.LocalTarget("sleepa_{taskid}_output.txt".format(taskid=self.taskid))

    def run(self):
        with self.output().open("w") as outfile:
            for i in xrange(60):
                time.sleep(1)
                outfile.write("Hej {i}\n".format(i=i))
                print "Wrote Hej to {outfile} with id {taskid} for the {i}st time".format(outfile=outfile.path, taskid=self.taskid, i=i)

class SleepB(luigi.Task):
    taskid = luigi.Parameter()

    def requires(self):
        return SleepA(self.taskid)

    def output(self):
        return luigi.LocalTarget("sleepb_{taskid}_output.txt".format(taskid=self.taskid))

    def run(self):
        with self.input().open() as infile, self.output().open("w") as outfile:
            i = 0 
            for line in infile:
                time.sleep(1)
                outfile.write(line)
                print "Wrote Hej to {outfile} with id {taskid} for the {i}st time".format(outfile=outfile.path, taskid=self.taskid, i=i)
                i += 1


class SleepC(luigi.Task):
    def requires(self):
        return [SleepB(taskid) for taskid in ["a", "b", "c"]]
    def output(self):
        return luigi.LocalTarget("sleepc_output.txt")
    def run(self):
        i = 0 
        with self.output().open("w") as outfile:
            for upstream_target in self.input():
                with upstream_target.open("r") as infile:
                    for line in infile:
                        time.sleep(1)
                        outfile.write(line)
                        print "Wrote line to {outfile} for the {i}st time".format(outfile=outfile.path, i=i)
                        i += 1


if __name__ == '__main__':
    luigi.run()

Run this on two nodes, say node1 and node2.

On both nodes, start the job with:

python multinode_luigi_workflow_test.py --scheduler-host node1 --workers 1 SleepC

... watch the output to make sure no two jobs run on both nodes.