一、luigi介绍
luigi是基于python语言的，可帮助建立复杂流式批处理任务管理系统。这些批处理作业典型的有hadoop job,数据库数据的导入与导出，或者是机器学习算法等等。
luigi的github：https://github.com/spotify/luigi
目前已经有一些抽象层次较低的数据处理工具，比如hive，pig，cascading等。luigi并不是要取代他们，而是帮助你管理这些作业，luigi的task可以是一个hive查询，java写的hadoop作业，一个scala写的spark作业或一个python程序等。luigi提供了互相依赖的大量作业的工作流程管理，所以程序员可以把他们的精力放到作业本身。
目前有一些相似的项目比如Oozie和Azkaban。一个重要的区别是luigi并不仅仅为hadoop作业，它可以很方便的扩展其他类型的任务。
二、luigi官网的hello world例子
2.1top Artists例子的目的
这个例子的目的想要集合一些生产数据的流，然后找到前10个artists，并把最终的结果保存到数据库
2.2Aggregate Artist Streams
class AggregateArtists(luigi.Task):
date_interval = luigi.DateIntervalParameter()
def output(self):
return luigi.LocalTarget("data/artist_streams_%s.tsv" % self.date_interval)
def requires(self):
return [Streams(date) for date in self.date_interval]
def run(self):
artist_count = defaultdict(int)
for input in self.input():
with input.open('r') as in_file:
for line in in_file:
timestamp, artist, track = line.strip().split()
artist_count[artist] += 1
with self.output().open('w') as out_file:
for artist, count in artist_count.iteritems():
print >> out_file, artist, count
对于这个类的解释：
requires方法：这个方法指定了本task需要的依赖，在这个例子中，AggregateArttists依赖一个Stream作业，Stream作业需要一个日期作为参数。
参数：每一个作业都可以定义一个或者多个参数，这些参数需要定义在类级别。比如上面这个类就有一个参数date_interval
output方法：定义了作业结果的保存地。
run方法：对于普通的task，你需要实现run方法。在run方法中可以是任何东西，可以创建子进程，进行长时间的算术运算等等。对于一些task的子类，你就不需要实现run方法了，比如JobTask要求你实现mapper和reducer方法。
LocalTarget:这是一个内置的类，可以帮助你很容易的读取或者写本地磁盘。并且保证对磁盘的操作是原子性的。
2.3Streams
class Streams(luigi.Task):
date = luigi.DateParameter()
def run(self):
with self.output().open('w') as output:
for _ in range(1000):
output.write('{} {} {}\n'.format(
random.randint(0, 999),
random.randint(0, 999),
random.randint(0, 999)))
def output(self):
return luigi.LocalTarget(self.date.strftime('data/streams_%Y_%m_%d_faked.tsv'))
这个类没有依赖，最终产生的效果是在本地文件系统上产生一个结果文件。
2.4在本地执行
PYTHONPATH='' luigi --module top_artists AggregateArtists --local-scheduler --date-interval 2012-06
执行完成之后，在当前目录下产生了一个data目录，data目录下的内容如下：
(my_python_env)[root@hadoop26 data]# ls artist_streams_2012-06.tsv streams_2012_06_06_faked.tsv streams_2012_06_12_faked.tsv streams_2012_06_18_faked.tsv streams_2012_06_24_faked.tsv streams_2012_06_30_faked.tsv streams_2012_06_01_faked.tsv streams_2012_06_07_faked.tsv streams_2012_06_13_faked.tsv streams_2012_06_19_faked.tsv streams_2012_06_25_faked.tsv streams_2012_06_02_faked.tsv streams_2012_06_08_faked.tsv streams_2012_06_14_faked.tsv streams_2012_06_20_faked.tsv streams_2012_06_26_faked.tsv streams_2012_06_03_faked.tsv streams_2012_06_09_faked.tsv streams_2012_06_15_faked.tsv streams_2012_06_21_faked.tsv streams_2012_06_27_faked.tsv streams_2012_06_04_faked.tsv streams_2012_06_10_faked.tsv streams_2012_06_16_faked.tsv streams_2012_06_22_faked.tsv streams_2012_06_28_faked.tsv streams_2012_06_05_faked.tsv streams_2012_06_11_faked.tsv streams_2012_06_17_faked.tsv streams_2012_06_23_faked.tsv streams_2012_06_29_faked.tsv
streams_*：就是stream作业生成的。
artist_*:是AggregateArtists生成的，就一个文件而已
2.5扩展
再次运行上面的执行命令发现并没有执行任何操作，因为所有任务的output已经存在。这意味着luigi的task都是幂等的，也就是说不管执行多少次，作业的输出应该是不变的。
--local-scheduler告诉luigi不要去连接scheduler server。这是不推荐的运行方式，这种方式也就用在测试阶段。