Pair RDDs: Transformations and Actions

Distributed key-value pairs are represented as Pair RDDs in Spark.

They are useful because they allow us to act on each key in parallel or regroup data across the network.

An RDD parameterized by a pair are treated as Pair RDD

RDD[(k, v)] // <------- treated specially by Spark

Pair RDDs can be created from already existing regular RDDs for example by using the mapoperation on the regular RDD:

val rdd: RDD[WikipediaPage] = ...
val pairRdd = rdd.map( wikipediapage => (wikipediapage.title, wikipediapage.text) )

Pair RDDs have additional, specialized transformation and actions operation for working with data associated with keys. Some of the commonly used are:

def groupByKey(): RDD[(K, Iterable[V])
def reduceByKey(func: (V, V) => V): RDD[(K, V)]
def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]

groupByKey (transformation)

groupByKey works on Pair RDDs, It groups value of the same key in an Iterable[V] of value.

def groupByKey(): RDD[(K, Iterable[V])]case class Event(Organizer: String, name: String, budget: Int)
val rdd = sc.parallelize(...)
val eventsRdd = rdd.map(event => (event.organizer, event.budget)) 
val groupedRDD = eventsRdd.groupByKey() 
groupedRDD.collect().foreach(println)Result :
// (Organizer1, CompactBuffer(42000))
// (Organizer2, CompactBuffer(20000, 44400, 87000))

reduceByKey (transformation)

It is a combination of groupByKey followed by reduce on values of each grouped collection. It is more efficient than using the both separately.

def reduceByKey( func(V, V) => V ): RDD[(K, V)] // V corresponds to the values of Pair RDD, we only operate on the value since a pair RDD is in the form of Key Values.calculate the total budget per organizationval eventsRdd = rdd.map(event => (event.organizer, event.budget)) 
val totalBudgetsRdd = eventsRdd.reduceByKey( _ + _ ) 
// at this point we already have keys and values. So reduceByKey means reduce the values corresponding to the given key using the given function.
 
totalBudgetsRdd.collect().foreach(println)
Result:// (Organizer1, 42000)
// (Organizer2, 151400)

mapValues (transformation)

It applies the given function to only the values in a Pair RDD i.e. transforms RDD[(K, V)] to RDD[(K, U)].

def mapValues[U](f: V => U): RDD[(K, U)]

sortByKey() (transformation)

It returns an RDD which is sorted by key. The result of sortByKey() is based on range-partitioned RDD.

RDD[(K, V)] to RDD[(K, V)]val rdd1 = sc.parallelize(Seq((“India”,91),(“USA”,1),(“Brazil”,55),(“Greece”,30),(“China”,86),(“Sweden”,46),(“Turkey”,90),(“Nepal”,977))) 
val rdd2 = rdd1.sortByKey()
rdd2.collect();output:Array[(String,Int)] = (Array(Brazil,55),(China,86),(Greece,30),(India,91),(Nepal,977),(Sweden,46),(Turkey,90),(USA,1)// may not print result in sorted order
rdd2.foreach(println)//will print result in sorted order
rdd2.collect().foreach(println)

foreach does not guarantee that the elements will be processed in any particular order. If you do sortByKeyRDD.collect.foreach(println) you will see the results in order, although this assumes that your data will fit in driver memory.

Calling collect or save on the resulting RDD will return or output an ordered list of records.

Why printing inside foreach doesn't reflect an order of elements

May be I am missing something but I expected the data to be sorted based on the key scala> val x=sc.parallelize(Array(…

stackoverflow.com

countByKey (action)

It counts the no. of elements per key in a Pair RDD and returns a regular Scala Map of the key against the count. It's an action so its eager.

def countByKey(): Map[K, Long]

Map vs FlatMap (transformation)

Map transforms one RDD into another using function provided and there is one to one mapping between the RDDs.

val rdd = sc.parallelize(Seq("Roses are red", "Violets are blue"))rdd.map(_.length).collect

    res1: Array[Int] = Array(13, 16)

FlatMap can generate many new rows from each row of rdd data. In flatMap function you pass in instead of returning single value it returns a list of values which contain many rows or maybe no rows at all. There is no one to one mapping between rows between the RDDs.

rdd.flatMap(_.split(" ")).collect

    res2: Array[String] = Array("Roses", "are", "red", "Violets", "are", "blue")