luigi_basic_pipeline_generate_words_and_count_lines.py

1import luigi
2
3class GenerateWords(luigi.Task):
4    def output(self):
5        return luigi.LocalTarget('words.txt')
6
7    def run(self):
8        # Write some words to a file
9        words = ['apple', 'banana', 'cherry', 'date']
10        with self.output().open('w') as f:
11            for word in words:
12                f.write(f'{word}\n')
13
14class CountWords(luigi.Task):
15    def requires(self):
16        return GenerateWords()
17
18    def output(self):
19        return luigi.LocalTarget('word_count.txt')
20
21    def run(self):
22        # Read the file from the previous task and count lines
23        with self.input().open('r') as f:
24            count = len(f.readlines())
25
26        with self.output().open('w') as f:
27            f.write(str(count))
28
29if __name__ == '__main__':
30    luigi.run()