莎士比亞文集詞頻統計並行化算法

你們好，很久沒沒更新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老楊，轉載請註明出處）