如何进行spark在日志分析系统里的简单应用

云计算

如何进行spark在日志分析系统里的简单应用 ,针对这个问题,这篇文章详细介绍了相对应的分析和解答,希望可以帮助更多想解决这个问题的小伙伴找到更简单易行的方法。

1.下载spark和运行

wget http://apache.fayea.com/apache-mirror/spark/spark-1.0.0/spark-1.0.0-bin-hadoop2.tgz

我这里下载的是1.0.0版,由于我们只是测试spark的用法所以不需要配置spark集群,只需把下好的文件解压,进入bin/文件夹。

spark支持scala,java和python。

scala和java输入命令:./spark-shell python 输入命令 ./pyspark 进入控制台,我们以python为例:

很简单无需任何配置,好的开始是成功的一半,我们继续。

2.简单应用

读取文本文件,我们还是以之前的日志文件为例:

>>>file = sc.textFile(“/home/hadoop/20130207.txt”) 

PS:python是大小写敏感的;路径要给全,否则默认是你运行命令时的那个路径;python2.7.X版本对编码支持一直是个问题,请尽量统一源文件的编码,如”utf-8″.

显示总行数:

>>> file.count()

265063

显示第一行内容:

>>>file.first()

得到所有被ie8访问的url的计数:

>>>file.filter(lambda line : “MSIE 8.0” in line).count()

98670

Ps: lambda 是匿名函数的写法;filter默认输入的是整行,上述代码的意思遍历每一行并计算出包含“MSIE 8.0”字符串的行数和。

得到拥有最多字段数的那行的字段个数:

>>> file.map(lambda line : len(line.split(“|”))).reduce(lambda a,b: a if a>b else b )

Ps:map(lambda line : len(line.split(“|”)))是把每行分割成集合并返回集合的成员数;

 python中的reduce内建函数是一个二元操作函数,他用来将一个数据集合(链表,元组等)中的所有数据进行下列操作:用传给reduce中的函数 func()(必须是一个二元操作函数)先对集合中的第1,2个数据进行操作,得到的结果再与第三个数据用func()函数运算,最后得到一个结果。

统计每个字符串(字段内容)出现的次数:

 file.flatMap(lambda line :line.split(“|”)).map(lambda word:(word,1)).reduceByKey(lambda a,b:a+b).collect()

结果显示的数据太多了,精神崩溃~~,换个方式~

统计每个字符串(字段内容)出现的次数,并显示出现最多个10个字符串:

 >>>file.flatMap(lambda line :line.split(“|”)).map(lambda word:(word,1)).reduceByKey(lambda a,b:a+b).map(lambda (k,v):(v,k)).sortByKey().top(10)

sortByKey是根据键值排序,top是提取前X个记录,跟hive中的limit 类似。由于spark中没有sortByValue方法,所以在排序前先把key和value互换。

关于如何进行spark在日志分析系统里的简单应用 问题的解答就分享到这里了,希望以上内容可以对大家有一定的帮助,如果你还有很多疑惑没有解开,可以关注箭头云行业资讯频道了解更多相关知识。

Published by

风君子

独自遨游何稽首 揭天掀地慰生平

发表回复

您的电子邮箱地址不会被公开。 必填项已用 * 标注