luigi框架–关于python运行spark程序

首先，目标是写个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 参数，选择你程序的运行队列，有时候默认的队列好像特别慢，可以设置个其他的。