莎士比亚文集词频统计并行化算法

你们好，很久没没更新Spark类容了，主要是最近考试比较多。今天先给你们展现一个实战案例，这个案例是我在今年参加第一届高校云计算应

用 创新大赛时技能赛第四题——莎士比亚文集词频统计并行化算加法。PS:感谢师兄辉哥，真大神!

题目是这样的（这里截图展现出来）：

a

在这里停词表的做用是对于在该表中的单词不予以统计，通常而言停词表中的单词是出现频率比较高的单词（如the）。这个案例比较简单，但

是要 好 优化仍是须要花费一点心思。

有的人的思路多是这样的：先对莎士比亚文集进行wordcount操做统计出各个单词的出现频率，而后对wordcount中的结果过滤掉在停词表

中出现的单词，最后找到出现频率最高的100个便可。这种方式可行，但效率略低。你们知道wordcount包含shuffle操做，shuffle所带来的IO是spark

性能的瓶颈。咱们在写程序的时候应该尽量的较少shuffle IO，那么如何减小shuffle IO呢，在这里咱们能够尽可能减小要参与shuffle操做的数据。

因此，优化的思路是对莎士比亚文集进行单词分片后就进行过滤操做，过滤掉在停词表中得单词，而后进行wordcount操做。这样一来咱们能够

过滤掉大量出现频率很高的词汇从而减小了主要shuffle IO。可能有的同窗会问那这里的filter操做岂不是比上面的思路中filter操做须要处理的单词书更

多，确实是这样。可是对性能没有任何影响，为何这么说？你们知道spark的一个优良的特色就是它的pipeline(血统)，咱们的处理在每个shuffle

操做之 前都会算做一个同一个stage,在这个satge中的计算都是在最后的action时才进行的，血统就是具备这一优良特性。那么对每个partiton上的

文本进行单词切割后进行filter操做是否是具备pipeline的特性？是否是这两个操做就像血液同样瞬间流过你的血管中的两个细胞？是否是几乎是同时发

生？是否是没有任何性能影响？

此外，咱们还能够将规模较小的停词表放在一个hash表中，hash查找的效率几乎为单位时间（你们必定要多关注hash的原理,前几天百度面试包

含了不少hash类容）。

说了这么多，下面贴出源码：

1. import org.apache.spark.SparkContext._
2. import org.apache.spark.SparkConf
3. import org.apache.spark._
4. 
   
5. object Shakespear {
6.   def main(args: Array[String]) {
7.     if (args.length != 3) {
8.       println("USage:<Shakespear> <Stopwords> <Out>")
9.     }
10.     //initial SparkConf and SparkContext
11.     val sc = new SparkContext()
12. 
    
13.     //To get Shakespear'paper
14.     val papers = sc.textFile(args(0))
15.     //To get stopwords
16.     val stopWords = sc.textFile(args(1)).map(_.trim).collect().toSet + ""
17. 
    
18.     //To parse papers into words and find the words statisfy the requirement
19.     val words = papers.flatMap(_.split("[^a-zA-Z]")).map(_.toLowerCase).filter(!stopWords(_)).map((_,1)).
20.                 reduceByKey(_ + _).map(line=>(line._2, line._1)).top(100).map(line=>(line._2, line._1))
21. 
    
22.     val result = sc.parallelize(words)
23. 
    
24.     //To write the result into hdfs
25.     result.saveAsTextFile(args(2))
26.   }
27. }

在后面我会提供包含技能赛第三题以及其余的案例详解。但愿你们共同窗习讨论。

（by老杨，转载请注明出处）