title: Spark: Efficient way to get top K frequent values per key in (key, value) RDD?
body: <p>I've an RDD of (key, value) pairs. I need to fetch top k values according to their frequencies for each key.</p>
  
  <p>I understand that the best way to do this would be using combineByKey. </p>
  
  <p>Currently here is what my combineByKey combinators look like</p>
  
  <pre><code>object TopKCount {
      //TopK Count combiners
      val k: Int = 10
      def createCombiner(value: String): Map[String, Long] = {
        Map(value -&gt; 1L)
      }
      def mergeValue(combined: Map[String, Long], value: String): Map[String, Long] = {
        combined ++ Map(value -&gt; (combined.getOrElse(value, 0L) + 1L))
      }
      def mergeCombiners(combined1: Map[String, Long], combined2: Map[String, Long]): Map[String, Long] = {
        val top10Keys1 = combined1.toList.sortBy(_._2).takeRight(k).toMap.keys
        val top10Keys2 = combined2.toList.sortBy(_._2).takeRight(k).toMap.keys
  
        (top10Keys1 ++ top10Keys2).map(key =&gt; (key, combined1.getOrElse(key, 0L) + combined2.getOrElse(key, 0L)))
          .toList.sortBy(_._2).takeRight(k).toMap
      }
    }
  </code></pre>
  
  <p>I use this as follows:</p>
  
  <pre><code>// input is RDD[(String, String)]
   val topKValueCount: RDD[(String, Map[String, Long])] = input.combineByKey(
        TopKCount.createCombiner,
        TopKCount.mergeValue,
        TopKCount.mergeCombiners
      )
  </code></pre>
  
  <p>One optimization to the current code would be to use <em>min-queue</em> during mergeCombiners. </p>
  
  <p>I'm more concerned about the network I/O. Would it be possible that once I do the merging in a Partition, I only send the topK entries from this partition to the driver, instead of sending the entire Map, which I'm doing in the current case.</p>
  
  <p>Highly appreciate any feedback.</p>