Setting Partitioner for RDD

When doing Join on Pair RDDs, if one of the dataset we are using is a Master data, it makes a lot of sense to persist the data, as we do not want the RDD being created every time an action associated with the dataset is executed.

Apart from enabling persistence for the master dataset, we can also avoid the shuffling of the master dataset by providing a Partitioner

Example

scala> val masterdata =
     |    sc.
     |    parallelize(Seq(
     |       ("math",    55),
     |       ("math",    56),
     |       ("english", 57),
     |       ("english", 58),
     |       ("science", 59),
     |       ("science", 54))).
     |    partitionBy(new HashPartitioner(100)).
     |    persist()
masterdata: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[42] at partitionBy at :31

scala> masterdata.partitioner
res16: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@64)

scala> masterdata.partitions.length
res17: Int = 100

Following are the operations, in which shuffling can be avoided/minimized by using a Partitioner
- cogroup()
- groupWith()
- join()
- leftOuterJoin()
- rightOuterJoin()
- groupByKey()
- reduceByKey()
- combineByKey()
- lookup()

Operations that preserves the Partitioner of the Parent RDD (or) result in Partitioner being set
- cogroup()
- groupWith()
- join()
- leftOuterJoin()
- rightOuterJoin()
- groupByKey()
- reduceByKey()
- combineByKey()
- partitionBy()
- sort()
- mapValues()

Operations that will not set a Partitioner
- map()

Reference

Learning Spark : Partitioning : 63, 66

groupBy() & groupByKey() Example

groupByKey() operates on Pair RDDs and is used to group all the values related to a given key. groupBy() can be used in both unpaired & paired RDDs. When used with unpaired data, the key for groupBy() is decided by the function literal passed to the method

Example

scala> val inputrdd = sc.parallelize(Seq(
     |                ("key1", 1),
     |                ("key2", 2),
     |                ("key1", 3)))
inputrdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[12] at parallelize at :21

//groupByKey() Example
scala> val grouped1 = inputrdd.groupByKey()
grouped1: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = ShuffledRDD[13] at groupByKey at :23

scala> grouped1.collect()
res6: Array[(String, Iterable[Int])] = Array((key1,CompactBuffer(1, 3)), (key2,CompactBuffer(2)))

//groupBy() Example : Find Odd & Even numbers
scala> val grouped2 = inputrdd.groupBy{ x =>
     |                      if((x._2 % 2) == 0) {
     |                         "evennumbers"
     |                      }else {
     |                         "oddnumbers"
     |                      }
     |                }
grouped2: org.apache.spark.rdd.RDD[(String, Iterable[(String, Int)])] = ShuffledRDD[15] at groupBy at :23

scala> grouped2.collect()
res7: Array[(String, Iterable[(String, Int)])] = Array((evennumbers,CompactBuffer((key2,2))), (oddnumbers,CompactBuffer((key1,1), (key1,3))))

Note

groupByKey() always results in Hash-Partitioned RDDs

Reference

Learning Spark : Hash-Partition : 64