title: Efficiently calculate top-k elements in spark
body: <p>I have a dataframe similarly to:</p>
  
  <pre><code>+---+-----+-----+
  |key|thing|value|
  +---+-----+-----+
  | u1|  foo|    1|
  | u1|  foo|    2|
  | u1|  bar|   10|
  | u2|  foo|   10|
  | u2|  foo|    2|
  | u2|  bar|   10|
  +---+-----+-----+
  </code></pre>
  
  <p>And want to get a result of:</p>
  
  <pre><code>+---+-----+---------+----+
  |key|thing|sum_value|rank|
  +---+-----+---------+----+
  | u1|  bar|       10|   1|
  | u1|  foo|        3|   2|
  | u2|  foo|       12|   1|
  | u2|  bar|       10|   2|
  +---+-----+---------+----+
  </code></pre>
  
  <p>Currently, there is code similarly to:</p>
  
  <pre><code>val df = Seq(("u1", "foo", 1), ("u1", "foo", 2), ("u1", "bar", 10), ("u2", "foo", 10), ("u2", "foo", 2), ("u2", "bar", 10)).toDF("key", "thing", "value")
  
   // calculate sums per key and thing
   val aggregated = df.groupBy("key", "thing").agg(sum("value").alias("sum_value"))
  
   // get topk items per key
   val k = lit(10)
   val topk = aggregated.withColumn("rank", rank over  Window.partitionBy("key").orderBy(desc("sum_value"))).filter('rank &lt; k)
  </code></pre>
  
  <p>However, this code is very <strong>inefficient</strong>. A window function generates a <strong>total order</strong> of items and causes a <strong>gigantic shuffle</strong>.</p>
  
  <p>How can I calculate top-k items more efficiently?
  Maybe using approximate functions i.e. sketches similarly to <a href="https://datasketches.github.io/" rel="nofollow noreferrer">https://datasketches.github.io/</a> or <a href="https://spark.apache.org/docs/latest/ml-frequent-pattern-mining.html" rel="nofollow noreferrer">https://spark.apache.org/docs/latest/ml-frequent-pattern-mining.html</a></p>