首页 > 代码库 > luigi框架--关于python运行spark程序
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 参数,选择你程序的运行队列,有时候默认的队列好像特别慢,可以设置个其他的。
luigi框架--关于python运行spark程序