Use combineByKey to get output as (key, iterable[v

Question

I am trying to transform RDD(key,value) to RDD(key,iterable[value]), same as output returned by the groupByKey method. But as groupByKey is not efficient, I am trying to use combineByKey on the RDD instead, however, it is not working. Below is the code used:

val data= List("abc,2017-10-04,15.2",
          "abc,2017-10-03,19.67", 
          "abc,2017-10-02,19.8",
          "xyz,2017-10-09,46.9", 
          "xyz,2017-10-08,48.4",
          "xyz,2017-10-07,87.5", 
          "xyz,2017-10-04,83.03", 
          "xyz,2017-10-03,83.41",
          "pqr,2017-09-30,18.18", 
          "pqr,2017-09-27,18.2", 
          "pqr,2017-09-26,19.2", 
          "pqr,2017-09-25,19.47", 
          "abc,2017-07-19,96.60",
          "abc,2017-07-18,91.68", 
          "abc,2017-07-17,91.55")
val rdd = sc.parallelize(templines)
val rows = rdd.map(line => {
  val row = line.split(",")
  ((row(0), row(1)), row(2))
})

// re partition and sort based key    
val op = rows.repartitionAndSortWithinPartitions(new CustomPartitioner(4))
val temp = op.map(f => (f._1._1, (f._1._2, f._2)))

val mergeCombiners = (t1: (String, List[String]), t2: (String, List[String])) => 
    (t1._1 + t2._1, t1._2.++(t2._2))
val mergeValue = (x: (String, List[String]), y: (String, String)) => {
  val a = x._2.+:(y._2)
  (x._1, a)
}

// createCombiner, mergeValue, mergeCombiners
val x = temp.combineByKey(
  (t1: String, t2: String) => (t1, List(t2)),
  mergeValue,
  mergeCombiners)

temp.combineByKey is giving compile time error, I am not able to get it.

Answer 1

If you want a output similar from what groupByKey will give you, then you should absolutely use groupByKey and not some other method. The reduceByKey, combineByKey, etc. are only more efficient compared to using groupByKey followed with an aggregation (giving you the same result as one of the other groupBy methods could have given).

As the wanted result is an RDD[key,iterable[value]], building the list yourself or letting groupByKey do it will result in the same amount of work. There is no need to reimplement groupByKey yourself. The problem with groupByKey is not its implementation but lies in the distributed architecture.

For more information regarding groupByKey and these types of optimizations, I would recommend reading more here.