Q:

How to optimize groupBy() operation on Spark RDD

My question is about alternatives/optimization to groupBy() operation on RDD. I have millions of Message instances which needs to be grouped based on some ID. I use groupBy() to accomplish the goal but I know the fact that it is  an expensive one and it takes huge amount of processing time .
So I tried reduceByKey(func) and combineByKey() as an alternative which was suggested in some of resources that I went through but its not a fit to my scenario where 
 My requirement is just grouping. But my data size is huge where groupBy consumes more than 70% of my job's time.So any suggestions or solution are appreciated.  
My input Data

Seq(new Message("1", "abc", "Request"), new Message("1", "cba",
  "Response"), new Message("2", "def", "Request"), new Message("2",
  "fed", "Response"), new Message("3", "ghi", "Request"), new
  Message("3", "ihg", "Response"))

For the sake of simplicity, i have given input data in Seq 
 def groupMessages(sourceRdd: RDD[Message]): Unit {

    val messageIdRdd = sourceRdd.mapPartitions(partitionData => {
      partitionData.map(row => (row.uniqueID(), row))

    })

    /*
     Group messages based on its message id. This functionality is to combine request response
     together as single transaction based on message id.
     */
    val groupbyRdd = messageIdRdd.groupBy(_._1) 
}

Expected Result 

(2,CompactBuffer((2,Message(2,def,Request)),
  (2,Message(2,fed,Response))))
  (3,CompactBuffer((3,Message(3,ghi,Request)),
  (3,Message(3,ihg,Response))))
  (1,CompactBuffer((1,Message(1,abc,Request)),
  (1,Message(1,cba,Response))))

A:

You should be using groupByKey() rather than groupBy() to have execution happen in cluster.
val myRDD = sc.parallelize( List(("1", "abc", "Request"), ("1", "cba", "Response"), ("2", "def", "Request"), ("2", "fed", "Response"), ("3", "ghi", "Request"), ("3", "ihg", "Response")) )

Your groupBy() is:
scala> myRDD.map(x=>x).groupBy(_._1).take(10).foreach(println)
(2,CompactBuffer((2,def,Request), (2,fed,Response)))
(3,CompactBuffer((3,ghi,Request), (3,ihg,Response)))
(1,CompactBuffer((1,abc,Request), (1,cba,Response))))

groupByKey() code would be:
scala> myRDD.map(x=>(x._1,x)).groupByKey().take(10).foreach(println)
(2,CompactBuffer((2,def,Request), (2,fed,Response)))
(3,CompactBuffer((3,ghi,Request), (3,ihg,Response)))
(1,CompactBuffer((1,abc,Request), (1,cba,Response))

Hope it helps.

