首先，目标是写个python脚本，跑spark程序来统计hdfs中的一些数据。参考了别人的代码，故用了luigi框架。
至于luigi的原理 底层的一些东西Google就好。本文主要就是聚焦快速使用，知其然不知其所以然。
python写Spark或mapreduce还有其他的方法，google上很多，这里用luigi只是刚好有参考的代码，而且理解起来还是简单，就用了。
上代码：
import luigi, sys
from datetime import datetime, timedelta
from luigi.contrib.spark import PySparkTask
class luigiBase(PySparkTask):
date = luigi.DateParameter(default=datetime.now())
def main(self, sc, *args):
log_rdd = sc.textFile(self.input()[0].path)
#要做的spark操作
log_rdd.repartition(1).saveAsTextFile(self.output().path)
@property
def name(self):
return "luigi_test_{}_username".format(format_date(self.date))
def requires(self):
return [HdfsFiles(date=self.date)]
def output(self):
return luigi.hdfs.HdfsTarget(Files().path,format=luigi.hdfs.PlainDir)
class luigiStats(luigi.Task):
now = datetime.now()
date = luigi.DateParameter(default=datetime(now.year, now.month, now.day) )
def requires(self):
return luigiBase(self.date)
if __name__ == '__main__':
luigi.run(main_task_cls=luigiStats)
1.对于普通的luigi任务，关键是要按需实现requires、output和run三个函数；对于luigi封装好的spark任务，关键是要按需实现requires、output和main三个函数
2.base类继承PySparkTask类，该类还有很多参数可以设置，但作为最最简单的luigi例子，就都剔除了，只要在意requires、output和main三个函数就好。可以把requires理解成输入，output输出，main是要实现的逻辑。name函数之所以也写出来，是因为在将代码pushonline的时候，每个Job都要取名字，而公司对job的名字是有规定的，如果name结尾不是你的用户名，Spark程序是会报错的，就是不让你跑的意思。
3.代码有两个类，base和stats类，执行逻辑是这样的：主函数调用stats，然后发现stats类requires（依赖于）base类，就看看这个依赖的输出存不存在，如果存在就作为自己的输入，然后执行自己类中的代码。如果不存在就执行base类。上面代码中我的stats类中不需要执行上面，就没写main，只是用来检查下base执行了没，没执行就执行base去。
3.该base类中requires和ouput都是hdfs文件，逻辑和stats类一样。base类需要继承PySparkTask类，而luigi.run()的参数需要时继承了luigi.Task的类，所以才分开写成两个类了，我自己是这样理解的。
4.requires函数的返回值不能是个target对象，这里具体的理解就是不能是一个直接读取的hdfs文件，可以封装到一个类中去，这个类可以有个属性是path,是用来返回一个hdfs文件的地址的。依赖不仅限一个，可以是多个，生成一个列表返回。
5.如果不是在自己的电脑上安装的Spark，要注意：由于PySparkTask调用的spark集群不在本地，好像不支持对本地文件的一些操作，开始的时候想把结果写在本地，一直找不到输出结果。
6.一般公司都有相对应得网页可以查看spark和hadoop程序的运行的情况，可以查看日志什么的。
7.base类中可以设置下queue 参数，选择你程序的运行队列，有时候默认的队列好像特别慢，可以设置个其他的。