Spark 編程實戰(zhàn)之經(jīng)典算法TOP K

Top K
Top K算法有兩步，一是統(tǒng)計詞頻，二是找出詞頻最高的前K個詞。
1.實例描述
假設(shè)取Top 1，則有如下輸入和輸出。
輸入：
Hello World Bye World
Hello Hadoop Bye Hadoop
Bye Hadoop Hello Hadoop
輸出：
詞Hadoop 詞頻4
2.設(shè)計思路
首先統(tǒng)計WordCount的詞頻，將數(shù)據(jù)轉(zhuǎn)化為（詞，詞頻）的數(shù)據(jù)對，第二個階段采用分
治的思想，求出RDD每個分區(qū)的Top K，最后將每個分區(qū)的Top K結(jié)果合并以產(chǎn)生新的集
合，在集合中統(tǒng)計出Top K的結(jié)果。每個分區(qū)由于存儲在單機的，所以可以采用單機求Top
K的方式。本例采用堆的方式。也可以直接維護一個含K個元素的數(shù)組，感興趣的讀者可以
參考其他資料了解堆的實現(xiàn)。
3.代碼示例
Top K算法示例代碼如下：
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
object TopK {
def main（args：Array[String]） {
/*執(zhí)行WordCount，統(tǒng)計出最高頻的詞*/
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（_ + _）
/*統(tǒng)計RDD每個分區(qū)內(nèi)的Top K查詢*/
val topk = count.mapPartitions（iter => {
while（iter.hasNext） {
putToHeap（iter.next（））
}
getHeap（）.iterator
}
）.collect（）
/*將每個分區(qū)內(nèi)統(tǒng)計出的TopK查詢合并為一個新的集合，統(tǒng)計出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）） {
/*數(shù)據(jù)加入含k個元素的堆中*/
……
}
def getHeap（）： Array[（String， Int）] = {
/*獲取含k個元素的堆中的元素*/
val a=new Array[（String， Int）]（）
……
}
4.應(yīng)用場景
Top K的示例模型可以應(yīng)用在求過去一段時間消費次數(shù)最多的消費者、訪問最頻繁的IP
地址和最近、更新、最頻繁的微博等應(yīng)用場景。