打造基于hadoop的网站日志分析系统（5）之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互换。