luigi框架--关于python运行spark程序
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了luigi框架--关于python运行spark程序,小编现在分享给大家,供广大互联网技能从业者学习和参考。文章包含2019字,纯文字阅读大概需要3分钟。
内容图文
首先,目标是写个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 参数,选择你程序的运行队列,有时候默认的队列好像特别慢,可以设置个其他的。
原文:http://www.cnblogs.com/qingjiaowoyc/p/6995097.html
内容总结
以上是互联网集市为您收集整理的luigi框架--关于python运行spark程序全部内容,希望文章能够帮你解决luigi框架--关于python运行spark程序所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错,欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 gblab@vip.qq.com 举报,一经查实,本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。