title: access each group in spark keyvaluegroupeddataset
body: <p>I am new to Spark and I have a specific question about how to use spark to address my problem, which may be simple.</p>
  <p><strong>Promblem:</strong></p>
  <p>I have a model, which predicts the sales of products. Each product also belongs to a category like shoes, clothes etc. And we also have actual sales data. So the data look like this:</p>
  <pre><code>+------------+----------+-----------------+--------------+
  | product_id | category | predicted_sales | actual_sales |
  +------------+----------+-----------------+--------------+
  |    pid1    |   shoes  |      100.0      |     123      |
  |    pid2    |   hat    |      232        |     332      |
  |    pid3    |   hat    |      202        |     432      |
  +------------+----------+-----------------+--------------+
  </code></pre>
  <p>product_id  category  predicted_sales  actual_sales</p>
  <p>What I'd like to do is: I want to calculate the number(or percentage) of intersection between top 5% products ranked by actual_sales and top 5% products ranked by predicted_sales for each category.</p>
  <p>Doing this for the whole products instead of for each category would be easy, something like below:</p>
  <pre><code>def getIntersectionRatio(df:dataframe, per :Int): Double = {
        val limit_num = (df.count() * per / 100.0).toInt
        var intersection = df.orderBy(&quot;actual_sales&quot;).limit(limit_num)
            .join(df.orderBy(&quot;predicted_sales&quot;).limit(limit_num), Seq(&quot;product_id&quot;), &quot;inner&quot;)
        intersection.count() * 100.0 / limit_num
  }
  </code></pre>
  <p>However, I need to calculate the intersection for each category. The result will be something like this</p>
  <pre><code>+-----------+------------------------+
  |  Category | intersection_percentage|
  +-----------+------------------------+
  </code></pre>
  <p><strong>My ideas</strong></p>
  <ol>
  <li>User Defined Aggreation Fuction or Aggregators
  I think I can achieve my goal if I use groupBy or GroupByKey with UDAF or Aggregators but they are too inefficient because they take 1 row each time and I will have to store each row in the buffer inside UDAF or Aggregator.</li>
  </ol>
  <pre><code>df.groupby(&quot;category&quot;).agg(myUdaf)
  
  class myUdaf extends UserDefinedAggregateFunction {
  //Save all the rows to an arraybuffer
  //and then transform the buffer back to df
  //And then we the the same thing as we did for whole product in  getIntersectionRatio defined previously
  }
  </code></pre>
  <ol start="2">
  <li>Self implemented partitioning
  I can select the distinct categories and the use map to process each category, in which I join the element with df to get the partition</li>
  </ol>
  <pre><code>df.select(&quot;category&quot;).distinct.map(myfun(df))
  
  def myfun(df: dataframe)(row : Row):Row = {
    val dfRow = row.toDF //not supported but feasible with other apis
    val group = df.join(broadcast(dfRow), seq(category), inner)
    getIntersectionRatio(group)
  }
  </code></pre>
  <p>Do we have a better solution for this?</p>
  <p>Thanks in advance!</p>