Simple Luigi task, useful for debugging server issue

simple_luigi_task.py

#!/usr/bin/env python

from luigi import build, Task, LocalTarget, FloatParameter
from luigi.util import requires
import numpy as np

class TaskA(Task):

    a= FloatParameter(default=round(np.random.rand(),3))

    def run(self):
        with open('/tmp/a.txt','w') as f:
            f.write('a')


    def output(self):
        return LocalTarget('/tmp/a.txt')

@requires(TaskA)
class TaskB(Task):

    b= FloatParameter(default=round(np.random.rand(),3))

    def run(self):
        with open('/tmp/b.txt', 'w') as f:
            f.write('b')

    def output(self):
        return LocalTarget('/tmp/b.txt')


@requires(TaskB)
class TaskC(Task):

    c= FloatParameter(default=round(np.random.rand(),3))

    def run(self):
        with open('/tmp/c.txt', 'w') as f:
            f.write('c')

    def output(self):
        return LocalTarget('/tmp/c.txt')


@requires(TaskC,TaskB)
class TaskD(Task):

    e= FloatParameter(default=round(np.random.rand(),3))

    def run(self):
        with open('/tmp/d.txt', 'w') as f:
            f.write('d')

    def output(self):
        return LocalTarget('/tmp/d.txt')


if __name__=='__main__':
    build([TaskD()])

Run instruction:

• kill any previous luigid

killall luigid

• obtain an IP, example

ip addr

• create luigi.cfg

[core]
default-scheduler-url = http://192.168.122.1:8082/

• restart luigid from the directory of luigi.cfg

luigid --address 192.168.122.1 --port 8082

• run the script

python simple_luigi_task.py