pip install bonobo

import bonobo

def generate_date():
    yield 'foo'
    yield 'bar'

def uppercase(x: str):
    return x.upper()

def output(x: str):
    print(x)

graph = bonobo.Graph(
    generate_date,
    uppercase,
    output
)

if __name__ == '__main__':
    bonobo.run(graph)

//Output
FOO
BAR
 - generate_date in=1 out=2 [done]
 - uppercase in=2 out=2 [done]
 - output in=2 [done]

pip install luigi

import luigi


class WritePipelineTask(luigi.Task):

    def output(self):
        return luigi.LocalTarget("data/output_one.txt")

    def run(self):
        with self.output().open("w") as output_file:
            output_file.write("pipeline")


class AddMyTask(luigi.Task):

    def output(self):
        return luigi.LocalTarget("data/output_two.txt")

    def requires(self):
        return WritePipelineTask()

    def run(self):
        with self.input().open("r") as input_file:
            line = input_file.read()

        with self.output().open("w") as output_file:
            decorated_line = "My "+line
            output_file.write(decorated_line)

# –local-scheduler 设置 luigi 不连接 scheduler 服务，仅建议在测试环境使用
$ PYTHONPATH='.' luigi --module luigi_demo AddMyTask --local-scheduler
DEBUG: Checking if AddMyTask() is complete
DEBUG: Checking if WritePipelineTask() is complete
INFO: Informed scheduler that task   AddMyTask__99914b932b   has status   PENDING
INFO: Informed scheduler that task   WritePipelineTask__99914b932b   has status   PENDING
INFO: Done scheduling tasks
INFO: Running Worker with 1 processes
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 2
INFO: [pid 1288] Worker Worker(salt=951591659, workers=1, host=zyt.local, username=zyt, pid=1288) running   WritePipelineTask()
INFO: [pid 1288] Worker Worker(salt=951591659, workers=1, host=zyt.local, username=zyt, pid=1288) done      WritePipelineTask()
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   WritePipelineTask__99914b932b   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
INFO: [pid 1288] Worker Worker(salt=951591659, workers=1, host=zyt.local, username=zyt, pid=1288) running   AddMyTask()
INFO: [pid 1288] Worker Worker(salt=951591659, workers=1, host=zyt.local, username=zyt, pid=1288) done      AddMyTask()
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   AddMyTask__99914b932b   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Done
DEBUG: There are no more tasks to run at this time
INFO: Worker Worker(salt=951591659, workers=1, host=zyt.local, username=zyt, pid=1288) was stopped. Shutting down Keep-Alive thread
INFO:
===== Luigi Execution Summary =====

Scheduled 2 tasks of which:
* 2 ran successfully:
    - 1 AddMyTask()
    - 1 WritePipelineTask()

This progress looks :) because there were no failed tasks or missing dependencies

===== Luigi Execution Summary =====

# 当前目录下出现一个 data 文件夹
$ ls
data luigi_demo.py

# 查看 data 文件夹
$ cd data
output_one.txt output_two.txt

# 分别查看文件内容
$ cat output_one.txt
pipeline

$ cat output_two.txt
My pipeline