luigi_two_task_pipeline_generate_text_count_lines.py

1import luigi
2
3class GenerateText(luigi.Task):
4    def output(self):
5        return luigi.LocalTarget("hello.txt")
6
7    def run(self):
8        with self.output().open("w") as f:
9            f.write("Hello world\n")
10            f.write("Luigi is working\n")
11
12class CountLines(luigi.Task):
13    def requires(self):
14        return GenerateText()
15
16    def output(self):
17        return luigi.LocalTarget("line_count.txt")
18
19    def run(self):
20        with self.input().open("r") as f:
21            lines = f.readlines()
22        
23        with self.output().open("w") as out:
24            out.write(str(len(lines)))
25
26if __name__ == '__main__':
27    luigi.run()