Spark: Efficient way to get top K frequent values per key in (key, value) RDD?
<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>