title: Spark aggregate on multiple columns within partition without shuffle
body: <p>I'm trying to aggregate a dataframe on multiple columns.  I know that everything I need for the aggregation is within the partition- that is, there's no need for a shuffle because all of the data for the aggregation are local to the partition.</p>
  
  <p>Taking an <a href="http://dmtolpeko.com/2015/02/12/multi-column-key-and-value-reduce-a-tuple-in-spark/" rel="noreferrer">example</a>, if I have something like</p>
  
  <pre><code>        val sales=sc.parallelize(List(
          ("West",  "Apple",  2.0, 10),
          ("West",  "Apple",  3.0, 15),
          ("West",  "Orange", 5.0, 15),
          ("South", "Orange", 3.0, 9),
          ("South", "Orange", 6.0, 18),
          ("East",  "Milk",   5.0, 5))).repartition(2)
          val tdf = sales.map{ case (store, prod, amt, units) =&gt; ((store, prod), (amt, amt, amt, units)) }.
          reduceByKey((x, y) =&gt; (x._1 + y._1, math.min(x._2, y._2), math.max(x._3, y._3), x._4 + y._4))
        println(tdf.toDebugString)
  </code></pre>
  
  <p>I get a result like</p>
  
  <pre><code>(2) ShuffledRDD[12] at reduceByKey at Test.scala:59 []
   +-(2) MapPartitionsRDD[11] at map at Test.scala:58 []
      |  MapPartitionsRDD[10] at repartition at Test.scala:57 []
      |  CoalescedRDD[9] at repartition at Test.scala:57 []
      |  ShuffledRDD[8] at repartition at Test.scala:57 []
      +-(1) MapPartitionsRDD[7] at repartition at Test.scala:57 []
         |  ParallelCollectionRDD[6] at parallelize at Test.scala:51 []
  </code></pre>
  
  <p>You can see the MapPartitionsRDD, which is good.  But then there's the ShuffleRDD, which I want to prevent because I want the per-partition summarization, grouped by column values within the partition.</p>
  
  <p><a href="https://stackoverflow.com/users/1560062/zero323">zero323</a>'s <a href="https://stackoverflow.com/questions/35154267/how-to-compute-cumulative-sum-using-spark">suggestion</a> is tantalizingly close, but I need the "group by columns" functionality.</p>
  
  <p>Referring to my sample above, I'm looking for the result that would be produced by</p>
  
  <pre><code>select store, prod, sum(amt), avg(units) from sales group by partition_id, store, prod
  </code></pre>
  
  <p>(I don't really need the partition id- that's just to illustrate that I want per-partition results)</p>
  
  <p>I've looked at <a href="https://stackoverflow.com/questions/43339027/sort-within-partition-but-not-sort-cross-the-partitions-using-spark-rdd">lots</a> <a href="https://github.com/jaceklaskowski/mastering-apache-spark-book/blob/master/spark-sql-performance-tuning-groupBy-aggregation.adoc" rel="noreferrer">of</a> <a href="https://stackoverflow.com/questions/33858750/sparksql-conditional-sum-using-two-columns">examples</a> but every debug string I've produced has the Shuffle.  I really hope to get rid of the shuffle.  I guess I'm essentially looking for a groupByKeysWithinPartitions function.</p>