Linked Luigi Example

linked_luigi.py

# Linked Luigi Example - Gouthaman Balaraman
# http://gouthamanbalaraman.com/blog/building-luigi-task-pipeline.html

import luigi

    
    
class SimpleTask(luigi.Task):
    """
    SimpleTask prints Hello World!.
    """
 
    def output(self):
        return MockFile("SimpleTask", mirror_on_stderr=True)
 
    def run(self):
        _out = self.output().open('w')
        _out.write(u"Hello World!\n")
        _out.close()

class DecoratedTask(luigi.Task):
    """
    DecoratedTask depends on the SimpleTask
    """
    
    def output(self):
        return MockFile("DecoratedTask", mirror_on_stderr=True)
    
    def requires(self):
        return SimpleTask()
    
    def run(self):
        _in = self.input().open("r")
        _out = self.output().open('w')
        for line in _in:
            outval = u"Decorated "+line+u"\n"
            _out.write(outval)
        
        _out.close()
        _in.close()
 
if __name__ == '__main__':
    from luigi.mock import MockFile # import this here for compatibility with Windows
    # if you are running windows, you would need --lock-pid-dir argument;
    # Modified run would look like
    # luigi.run(["--lock-pid-dir", "D:\\temp\\", "--local-scheduler"], main_task_cls=DecoratedTask)
    luigi.run(["--local-scheduler"], main_task_cls=DecoratedTask)