您的位置：首页 > 编程语言

一个Spark Demo与代码详解

2017-08-04 18:01 218 查看
简单的Nginx日志统计

代码详解
##载入依赖包

from pyspark import SparkContext		 
##生成并初始化一个Spark任务

sc = SparkContext("local", "Simple App")		SparkContext($1, $2)

$1：指定工作方式。

  可选项：

    local：默认值，使用一个 Worker 线程本地化运行(完全不并行)

    local
：使用 N 个 Worker 线程本地化运行，N 为 * 时，表示使用系统中所有核

    local[N,M]：N代表的是用到的核个数；M代表的是容许该作业失败M次

$2：指定任务名称。

其他初始方法：

from pyspark import SparkContext, SparkConf

conf = SparkConf().setAppName(APP_NAME)

conf = conf.setMaster("local[*]")

sc = SparkContext(conf=conf)
##载入一个外部文件

logFile = "access.log"

rdd = sc.textFile(logFile).cache()		cache：RDD持久化操作。
##进行统计

counts = rdd.map(lambda line: line.split()[8]).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b).sortByKey(True)		lambda：python定义匿名函数的标识符（详情见：python匿名函数.note）

map：spark转换之一，逐条将RDD中的元素（对于文本文件默认一行就是一个元素）使用传入的func进行处理。类比shell中的cat xxx | awk ‘{func}’

reduceByKey：spark转换之一，依据key聚合。key即是第二个map返回的元组的第一个值。参数a是当前聚合结果，参数b是下一项的value，即第二个map返回的元组的第二个值。

sortByKey：spark转换之一，依据key排序，传入的True表示由小到大
##输出结果

output = counts.collect()

for (word, count) in output:

print "%s: %i" % (word, count)		collect：以数组的形式，返回数据集的所有元素。
##将结果保存到文件

counts.saveAsTextFile("/data/result")		saveAsTextFile：将数据集的元素，以 textfile 的形式，保存到文件系统中
##停止Spark任务，释放资源

sc.stop()		 
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 点击这里给我发消息
标签： 
相关文章推荐
新的分享
章节导航