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

parallelize() Examples

scala> //for loop Example
scala> val rdd = sc.parallelize(for {
     |     x &lt- 1 to 3
     |     y &lt- 1 to 2
     | } yield (x, None))
rdd: org.apache.spark.rdd.RDD[(Int, None.type)] = ParallelCollectionRDD[38] at parallelize at :21

scala>

scala> rdd.collect()
res15: Array[(Int, None.type)] = Array((1,None), (1,None), (2,None), (2,None), (3,None), (3,None))

Reference

for loop with yield : http://alvinalexander.com/scala/scala-for-loop-yield-examples-yield-tutorial

Actions on Pair RDDs : countByKey(), collectAsMap() & lookup()

Example

scala> val rdd = sc.parallelize(Seq(
     |                ("math",    55),
     |                ("math",    56),
     |                ("english", 57),
     |                ("english", 58),
     |                ("science", 59),
     |                ("science", 54)))
rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[31] at parallelize at :21

scala> //Example : countByKey()
scala> val result1 = rdd.countByKey()
result1: scala.collection.Map[String,Long] = Map(math -> 2, english -> 2, science -> 2)

scala> //Example : collectAsMap()
scala> val reslt2 = rdd.collectAsMap()
reslt2: scala.collection.Map[String,Int] = Map(math -> 56, science -> 54, english -> 58)

scala> //Example : lookup()
scala> val result3 = rdd.lookup("math")
result3: Seq[Int] = WrappedArray(55, 56)

sortByKey() Example

sortByKey() is part of OrderedRDDFunctions that works on Key/Value pairs. The official documentation for OrderedRDDFunctions states that,

class OrderedRDDFunctions[K, V, P &lt: Product2[K, V]] extends Logging with Serializable
Extra functions available on RDDs of (key, value) pairs where the key is sortable through an implicit conversion. They will work with any key type K that has an implicit Ordering[K] in scope. Ordering objects already exist for all of the standard primitive types. Users can also define their own orderings for custom types, or to override the default ordering. The implicit ordering that is in the closest scope will be used.

Syntax : sortByKey()

def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)]
Sort the RDD by key, so that each partition contains a sorted range of the elements. Calling collect or save on the resulting RDD will return or output an ordered list of records (in the save case, they will be written to multiple part-X files in the filesystem, in order of the keys).

Example

scala> //Input
scala> val rdd = sc.parallelize(Seq(
     |                ("math",    55),
     |                ("math",    56),
     |                ("english", 57),
     |                ("english", 58),
     |                ("science", 59),
     |                ("science", 54)))
rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[17] at parallelize at :21

scala> rdd.collect()
res8: Array[(String, Int)] = Array((math,55), (math,56), (english,57), (english,58), (science,59), (science,54))

scala> //Default Sorting : Ascending order
scala> val sorted1 = rdd.sortByKey()
sorted1: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[20] at sortByKey at :23

scala> //Result
scala> sorted1.collect()
res9: Array[(String, Int)] = Array((english,57), (english,58), (math,55), (math,56), (science,59), (science,54))

scala> //Custom Sorting : Descending order (using implicit 'Ordering')
scala> {
     |    //Let us define an implicit sorting for the method sortByKey()
     |    //We have used '{' above to limit the scope of the implicit ordering
     |    implicit val sortIntegersByString = new Ordering[String] {
     |       override def compare(a: String, b: String) = {
     |          val result = a.compare(b)
     |          //We use -ve to sort the key in descending order
     |          -result
     |       }
     |    }
     |    val sorted2 = rdd.sortByKey()
     |
     |    //Result
     |    sorted2.collect()
     | }
res10: Array[(String, Int)] = Array((science,59), (science,54), (math,55), (math,56), (english,57), (english,58))

scala> //Default Sorting : Descending order (done using the 'ascending' flag argument)
scala> val sorted3 = rdd.sortByKey(false)
sorted3: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[26] at sortByKey at :23

scala> //Result
scala> sorted3.collect()
res11: Array[(String, Int)] = Array((science,59), (science,54), (math,55), (math,56), (english,57), (english,58))

Note : sortByKey() results in range-partitioned RDDs

Reference

Learning Spark : Range Partition : 64

join(), leftOuterJoin() & rightOuterJoin() Example

Example

scala> //Input Data
scala> val rdd1 = sc.parallelize(Seq(
     |                ("math",    55),
     |                ("math",    56),
     |                ("english", 57),
     |                ("english", 58),
     |                ("science", 59),
     |                ("science", 54)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[45] at parallelize at :21

scala> val rdd2 = sc.parallelize(Seq(
     |                ("math",    60),
     |                ("math",    65),
     |                ("science", 61),
     |                ("science", 62),
     |                ("history", 63),
     |                ("history", 64)))
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[46] at parallelize at :21

scala> //join() Example
scala> val joined = rdd1.join(rdd2)
joined: org.apache.spark.rdd.RDD[(String, (Int, Int))] = MapPartitionsRDD[49] at join at :25

scala> //Result
scala> joined.collect()
res15: Array[(String, (Int, Int))] = Array((math,(55,60)), (math,(55,65)), (math,(56,60)), (math,(56,65)), (science,(59,61)), (science,(59,62)), (science,(54,61)), (science,(54,62)))

scala> //leftOuterJoin() Example
scala> val leftJoined = rdd1.leftOuterJoin(rdd2)
leftJoined: org.apache.spark.rdd.RDD[(String, (Int, Option[Int]))] = MapPartitionsRDD[52] at leftOuterJoin at :25

scala> //Result
scala> leftJoined.collect()
res16: Array[(String, (Int, Option[Int]))] = Array((math,(55,Some(60))), (math,(55,Some(65))), (math,(56,Some(60))), (math,(56,Some(65))), (english,(57,None)), (english,(58,None)), (science,(59,Some(61))), (science,(59,Some(62))), (science,(54,Some(61))), (science,(54,Some(62))))

scala> //rightOuterJoin() Example
scala> val rightJoined = rdd1.rightOuterJoin(rdd2)
rightJoined: org.apache.spark.rdd.RDD[(String, (Option[Int], Int))] = MapPartitionsRDD[55] at rightOuterJoin at :25

scala> //Result
scala> rightJoined.collect()
res17: Array[(String, (Option[Int], Int))] = Array((math,(Some(55),60)), (math,(Some(55),65)), (math,(Some(56),60)), (math,(Some(56),65)), (history,(None,63)), (history,(None,64)), (science,(Some(59),61)), (science,(Some(59),62)), (science,(Some(54),61)), (science,(Some(54),62)))