mapValues() Example

When we use map() with a Pair RDD, we get access to both Key & value. There are times we might only be interested in accessing the value(& not key). In those case, we can use mapValues() instead of map().

In this example we use mapValues() along with reduceByKey() to calculate average for each subject

scala> val inputrdd = sc.parallelize(Seq(("maths", 50), ("maths", 60), ("english", 65)))
inputrdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[29] at parallelize at :21

scala> val mapped = inputrdd.mapValues(mark => (mark, 1));
mapped: org.apache.spark.rdd.RDD[(String, (Int, Int))] = MapPartitionsRDD[30] at mapValues at :23

scala> val reduced = mapped.reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))
reduced: org.apache.spark.rdd.RDD[(String, (Int, Int))] = ShuffledRDD[31] at reduceByKey at :25

scala> val average = reduced.map { x =>
     |                      val temp = x._2
     |                      val total = temp._1
     |                      val count = temp._2
     |                      (x._1, total / count)
     |                      }
average: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[32] at map at :27

scala>
     | average.collect()
res30: Array[(String, Int)] = Array((english,65), (maths,55))

Note 

Operations like map() always cause the new RDD to no retain the parent partitioning information

Reference

Learning Spark : Partitioning : 64

filter(), collect() & more...

In this post we demonstrate the following methods,

1. Transformation 
• filter()
2. Actions
• take()
• collect()
• saveAsTextFile()

scala> val inputRDD = sc.parallelize(1 to 10)
inputRDD: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[9] at parallelize at :21

scala> //Filter number '5'
scala> val filteredRDD = inputRDD.filter(x => x != 5)
filteredRDD: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[10] at filter at :23

scala> //Collect all the records
scala> //Warning : This will collect the entire data in RDD
scala> //          into local machine memory
scala> filteredRDD.collect()
res15: Array[Int] = Array(1, 2, 3, 4, 6, 7, 8, 9, 10)

scala> //Collect only 2 records from the RDD into Local Machine meory
scala> filteredRDD.take(2)
res16: Array[Int] = Array(1, 2)

scala> //Save the result to text file
scala> filteredRDD.saveAsTextFile("./result")

cat result/*
1
2
3
4
6
7
8
9
10