一，介绍
Luigi 是一个 Python 模块，可以帮你构建复杂的批量作业管道。处理依赖决议、工作流管理、可视化展示等等，内建 Hadoop 支持。它也被Foursquare，Stripe，华尔街日报，Groupon和其他知名企业使用。
Luigi是基于代码的，而不是基于GUI或声明式的，包含Python中的所有内容（包括依赖关系图）。用户界面（UI）允许您搜索，过滤或监视每个任务的状态。您还可以查看该工作流程，以查看依赖关系图上的哪些任务已完成，哪些尚未运行。
Luigi轻便简洁， 能够快速实现并行化的工作流。
二，安装和基本概念
1，安装
pip install luigi
2, 基本概念
luigi 有三个重要概念，Task, Target 和parameter
Task
每一个任务都是一个Task，以class的形式存在，继承luigi.Task。需要重载requires()、run()、output()方法。
其中requires()是任务入口程序，指定任务依赖的上游输入；run()是任务在该节点具体实现的流程；output()是任务的出口，把该节点执行完之后的结果输出到下游。
Target
广义地讲，Target可对应为磁盘上的文件，或HDFS上文件,或checkpoint点，或数据库等。对于Target来说，唯一需要实现的方法为exists,返回为True表示存在，否则不存在返回为False. 在实际应用时，写一个Target子类是很少需要用到的。直接使用开箱即可用的LocalTarget及 hdfs.HdfsTarget类就够用了。Luigi提供了Gzip支持，通过参数format=format.Gzip即可。
自定义Target的示例
# 实现判断数据库中test_table表是否被创建的Target
import luigi
import MySQLdb
class ConnectDB:
def __init__(self, host="128.0.0.1", user="root", passwd="root", db="test", port=3306):
self.conn = MySQLdb.connect(host, user, passwd, db, port=port, charset='utf8')
def __enter__(self):
return self.conn
def __exit__(self, exc_type, exc_val, exc_tb):
self.conn.close()
def connect(self):
return self.conn
class MysqlTarget(luigi.Target):
def __init__(self, primer_table):
self.primer_table = primer_table
def exists(self):
with ConnectDB() as conn:
sql = "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = '{}'".format(self.test_table)
cur = conn.cursor()
cur.execute(sql)
table = cur.fetchone()
cur.close()
return True if table else False
parameter
parameter等效于luigi为task类创建构造函数，Luigi中提供了不同类型的parameter，例如DateParameter，DateIntervalParameter，IntParameter,FloatParameter等等。
python不是一个静态类型的语言，你不需要指定参数的类型，你可以直接使用基类Parameter。
class TaskA(luigi.Task):
x = luigi.Parameter()
y = luigi.IntParameter(default=0)
三，示例代码
在这个例子中我们定义PrintNum任务创建一串数字写入到文件中，其他平方SquareNum，求倍数AddNum和立方CubeNum任务依赖这个文件，把自己的计算结果写如到自己对应的文件中。
#!/usr/bin/python
# _*_ coding:utf-8 _*_
import luigi
import time
import uuid
UID = str(uuid.uuid4())
# 生成1， n个数字，写入到number_up_{}.txt中
class PrintNum(luigi.Task):
n = luigi.IntParameter(default=5)
lt = list()
def requires(self):
return []
def output(self):
return luigi.LocalTarget("number_up_{}.txt".format(UID))
def run(self):
with self.output().open("w") as f:
for i in range(1, self.n):
self.set_status_message("Progress: %d" % i)
self.set_progress_percentage(i)
f.write("{}\n".format(i))
# 依赖PrintNum，对数字求平方后写入到squares_{}.txt中
class SquareNum(luigi.Task):
n = luigi.IntParameter(default=10)
def requires(self):
return [PrintNum()]
def output(self):
return luigi.LocalTarget("squares_{}.txt".format(UID))
def run(self):
with self.input()[0].open() as fin, self.output().open("w") as fout:
for line in fin:
n = int(line.strip())
list(range(10**7))
# time.sleep(1)
out = n * n
fout.write("{}:{}\n".format(n, out))
# 依赖PrintNum，对数字求倍数后写入到add_{}.txt中
class AddNum(luigi.Task):
n = luigi.IntParameter(default=10)
def requires(self):
return [PrintNum(n=self.n)]
def output(self):
return luigi.LocalTarget("add_{}.txt".format(UID))
def run(self):
with self.input()[0].open() as fin, self.output().open("w") as fout:
for line in fin:
n = int(line.strip())
list(range(10**7))
# time.sleep(1)
out = n + n
fout.write("{}:{}\n".format(n, out))
# 依赖PrintNum，对数字求立方后写入到cube_{}.txt中
class CubeNum(luigi.Task):
n = luigi.IntParameter(default=10)
def requires(self):
return[PrintNum(n=self.n)]
def output(self):
return luigi.LocalTarget("cube_{}.txt".format(UID))
def run(self):
with self.input()[0].open() as fin, self.output().open("w") as fout:
for line in fin:
n = int(line.strip())
list(range(10**7))
# time.sleep(1)
out = n ** 3
fout.write("{}:{}\n".format(n, out))
if __name__ == "__main__":
#构建任务
luigi.build([SquareNum(n=40), AddNum(n=40), CubeNum(n=40)], local_scheduler=True)
