Using reduceByKey in Apache Spark (Scala)

Question

I have a list of Tuples of type : (user id, name, count).

For example,

val x = sc.parallelize(List(     ("a", "b", 1),     ("a", "b", 1),     ("c", "b", 1),     ("a", "d", 1)) ) 

I'm attempting to reduce this collection to a type where each element name is counted.

So in above val x is converted to :

(a,ArrayBuffer((d,1), (b,2))) (c,ArrayBuffer((b,1))) 

Here is the code I am currently using :

val byKey = x.map({case (id,uri,count) => (id,uri)->count})  val grouped = byKey.groupByKey val count = grouped.map{case ((id,uri),count) => ((id),(uri,count.sum))} val grouped2: org.apache.spark.rdd.RDD[(String, Seq[(String, Int)])] = count.groupByKey  grouped2.foreach(println) 

I'm attempting to use reduceByKey as it performs faster than groupByKey.

How can reduceByKey be implemented instead of above code to provide the same mapping ?

Answer 1

Following your code:

val byKey = x.map({case (id,uri,count) => (id,uri)->count}) 

You could do:

val reducedByKey = byKey.reduceByKey(_ + _)  scala> reducedByKey.collect.foreach(println) ((a,d),1) ((a,b),2) ((c,b),1) 

PairRDDFunctions[K,V].reduceByKey takes an associative reduce function that can be applied to the to type V of the RDD[(K,V)]. In other words, you need a function f[V](e1:V, e2:V) : V . In this particular case with sum on Ints: (x:Int, y:Int) => x+y or _ + _ in short underscore notation.

For the record: reduceByKey performs better than groupByKey because it attemps to apply the reduce function locally before the shuffle/reduce phase. groupByKey will force a shuffle of all elements before grouping.