TOP K, the Classic algorithm of Spark programming

The Top KTop K algorithm has two steps: one is to count the word frequency, and the other is to find out the first K words with the highest word frequency. 1. If you take Top 1 as the example description, you will have the following inputs and outputs. Input: Hello World Bye WorldHello HadoopBye HadoopBye Hadoop Hello Hadoop output: word Hadoop word frequency 42. The design idea first counts the word frequency of WordCount and converts the data into data pairs (word, word frequency). In the second stage, the idea of divide and conquer is used to find out the Top K of each partition of RDD. Finally, the Top K results of each partition are merged to produce a new set, and the results of Top K are counted in the set. Because each partition is stored in a stand-alone machine, you can use a single machine to find the TopK. This example uses a heap approach. You can also directly maintain an array of K elements, and interested readers can refer to other materials to learn about the implementation of the heap. 3. Code example TopK algorithm example code is as follows: import org.apache.spark.SparkContextimport org.apache.spark.SparkContext._object TopK {def main (args: array [string]) {/ * execute WordCount, count out the most frequent word * / 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) ReduceByKey (_ + _) / * Statistics TopK queries within each partition of RDD * / val topk = count.mapPartitions (iter = > {while (iter.hasNext) {putToHeap (iter.next ())} getHeap (). Iterator}). Collect () / * merges the TopK queries counted within each partition into a new collection Statistics show that TopK query * / val iter= topk.iteratorwhile (iter.hasNext) {putToHeap (iter.next ())} val outiter=getHeap () .iterator / * output TopK value * / println ("Topk value:") while (outiter.hasNext) {println ("\ n word frequency:" + outiter.next (). _ 1 + "word:" + outiter.next (). _ 2)} spark.stop ()} def putToHeap (iter: (String) Int) {/ * data is added to the heap with k elements * /... } def getHeap (): Array [(String, Int)] = {/ * get the element * / val a=new Array [(String, Int)] () in the heap with k elements. } 4. The example model of application scenario Top K can be used to find the most frequently consumed consumers, the most frequently visited IP addresses and the most recent, updated and most frequently Weibo application scenarios in the past.