Spark 編程實戰之經典算法TOP K

Top K
Top K算法有兩步，一是統計詞頻，二是找出詞頻最高的前K個詞。
1.實例描述
假設取Top 1，則有如下輸入和輸出。
輸入：
Hello World Bye World
Hello Hadoop Bye Hadoop
Bye Hadoop Hello Hadoop
輸出：
詞Hadoop 詞頻4
2.設計思路
首先統計WordCount的詞頻，將數據轉化為（詞，詞頻）的數據對，第二個階段采用分
治的思想，求出RDD每個分區的Top K，最后將每個分區的Top K結果合并以產生新的集
合，在集合中統計出Top K的結果。每個分區由于存儲在單機的，所以可以采用單機求Top
K的方式。本例采用堆的方式。也可以直接維護一個含K個元素的數組，感興趣的讀者可以
參考其他資料了解堆的實現。
3.代碼示例
Top K算法示例代碼如下：
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
object TopK {
def main（args：Array[String]） {
/*執行WordCount，統計出最高頻的詞*/
val spark = new SparkContext（"local"， "TopK"，
System.getenv（"SPARK_HOME"）， SparkContext.jarOfClass（this.getClass））
val count = spark.textFile（"data"）.flatMap（line =>
line.split（" "））.map（word =>
（word， 1））.reduceByKey（_ + _）
/*統計RDD每個分區內的Top K查詢*/
val topk = count.mapPartitions（iter => {
while（iter.hasNext） {
putToHeap（iter.next（））
}
getHeap（）.iterator
}
）.collect（）
/*將每個分區內統計出的TopK查詢合并為一個新的集合，統計出TopK查詢*/
val iter = topk.iterator
while（iter.hasNext） {
putToHeap（iter.next（））
}
val outiter=getHeap（）.iterator
/*輸出TopK的值*/
println（"Topk 值 ："）
while（outiter.hasNext） {
println（"\n 詞頻："+outiter.next（）._1+" 詞："+outiter.next（）._2）
}
spark.stop（）
}
}
def putToHeap（iter ： （String， Int）） {
/*數據加入含k個元素的堆中*/
……
}
def getHeap（）： Array[（String， Int）] = {
/*獲取含k個元素的堆中的元素*/
val a=new Array[（String， Int）]（）
……
}
4.應用場景
Top K的示例模型可以應用在求過去一段時間消費次數最多的消費者、訪問最頻繁的IP
地址和最近、更新、最頻繁的微博等應用場景。