A Word Count example using 'spark-shell'

[raj@Rajkumars-MacBook-Pro ~]$spark-shell --master local[*] 
2016-05-28 15:37:24.325 java[3907:6309927] Unable to load realm info from SCDynamicStore
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 1.6.1
      /_/

Using Scala version 2.10.5 (Java HotSpot(TM) 64-Bit Server VM, Java 1.7.0_45)
Type in expressions to have them evaluated.
Type :help for more information.
Spark context available as sc.
SQL context available as sqlContext.

scala> val lines = sc.parallelize(List("This is a word", "This is another word"), 7)
lines: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at :27

scala> // No of partitions

scala> lines.partitions.size
res0: Int = 7

scala> // flatMap() : One of many transformation

scala> val words = lines.flatMap(line => line.split(" "))
words: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[1] at flatMap at :29

scala> val units = words.map ( word => (word, 1) )
units: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[2] at map at :31

scala> val counts = units.reduceByKey ( (x, y) => x + y )
counts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[3] at reduceByKey at :33

scala> counts.toDebugString
res1: String = 
(7) ShuffledRDD[3] at reduceByKey at :33 []
 +-(7) MapPartitionsRDD[2] at map at :31 []
    |  MapPartitionsRDD[1] at flatMap at :29 []
    |  ParallelCollectionRDD[0] at parallelize at :27 []

scala> // collect() : One of many actions

scala> counts.collect()
res2: Array[(String, Int)] = Array((This,2), (is,2), (another,1), (a,1), (word,2))

Combiner in Pair RDDs : combineByKey()

Similar to combiner in MapReduce, when working with key/value pairs, combineByKey() interface can be used to customize the combiner functionality. Methods like reduceByKey() by default use their own combiner to combine the data locally in each Partition, for a given key

Similar to aggregate()(which is used with single element RDD), combineByKey() allows user to return different RDD element type compared to the element type of Input RDD

Syntax

def combineByKey[C](createCombiner: (V) ⇒ C, mergeValue: (C, V) ⇒ C, mergeCombiners: (C, C) ⇒ C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = null): RDD[(K, C)]
Generic function to combine the elements for each key using a custom set of aggregation functions. Turns an RDD[(K, V)] into a result of type RDD[(K, C)], for a "combined type" C Note that V and C can be different -- for example, one might group an RDD of type (Int, Int) into an RDD of type (Int, Seq[Int]). Users provide three functions:

- createCombiner, which turns a V into a C (e.g., creates a one-element list) - mergeValue, to merge a V into a C (e.g., adds it to the end of a list) - mergeCombiners, to combine two C's into a single one.

In addition, users can control the partitioning of the output RDD, and whether to perform map-side aggregation (if a mapper can produce multiple items with the same key).

Here,
1st Argument : createCombiner is called when a key(in the RDD element) is found for the first time in a given Partition. This method creates an initial value for the accumulator for that key
2nd Argument : mergeValue is called when the key already has an accumulator
3rd Argument : mergeCombiners is called when more that one partition has accumulator for the same key

Example

Let us calculate the average in each subject using combineByKey()

scala> val inputrdd = sc.parallelize(Seq(
     |                    ("maths", 50), ("maths", 60),
     |                    ("english", 65),
     |                    ("physics", 66), ("physics", 61), ("physics", 87)), 
     |                    1)
inputrdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[41] at parallelize at <console>:27

scala> inputrdd.getNumPartitions                      
res55: Int = 1

scala> val reduced = inputrdd.combineByKey(
     |     (mark) => {
     |       println(s"Create combiner -> ${mark}")
     |       (mark, 1)
     |     },
     |     (acc: (Int, Int), v) => {
     |       println(s"""Merge value : (${acc._1} + ${v}, ${acc._2} + 1)""")
     |       (acc._1 + v, acc._2 + 1)
     |     },
     |     (acc1: (Int, Int), acc2: (Int, Int)) => {
     |       println(s"""Merge Combiner : (${acc1._1} + ${acc2._1}, ${acc1._2} + ${acc2._2})""")
     |       (acc1._1 + acc2._1, acc1._2 + acc2._2)
     |     }
     | )
reduced: org.apache.spark.rdd.RDD[(String, (Int, Int))] = ShuffledRDD[42] at combineByKey at <console>:29

scala> reduced.collect()
Create combiner -> 50
Merge value : (50 + 60, 1 + 1)
Create combiner -> 65
Create combiner -> 66
Merge value : (66 + 61, 1 + 1)
Merge value : (127 + 87, 2 + 1)
res56: Array[(String, (Int, Int))] = Array((maths,(110,2)), (physics,(214,3)), (english,(65,1)))

scala> val result = reduced.mapValues(x => x._1 / x._2.toFloat)
result: org.apache.spark.rdd.RDD[(String, Float)] = MapPartitionsRDD[43] at mapValues at <console>:31

scala> result.collect()
res57: Array[(String, Float)] = Array((maths,55.0), (physics,71.333336), (english,65.0))

Note

The map side aggregation done using combineByKey() can also be disabled(which is the case with methods like groupByKey() where the functionality of the combiner is not needed)

Reference

Learning Spark : Page 54

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

Pair RDD generation

Using parallelize() method

By creating Tuples using the parallelize() method, Pair RDD can be created

scala> val inputrdd = sc.parallelize(Seq(("key1", "val1"), ("key1", "val1")))
inputrdd: org.apache.spark.rdd.RDD[(String, String)] = ParallelCollectionRDD[3] at parallelize at :21

scala> inputrdd.reduceByKey((x, y) => x.concat(y)).collect()
res0: Array[(String, String)] = Array((key1,val1val1))

Note :
reduceByKey() is available in class PairRDDFunctions

Pair RDDs are simply RDDs of Tuple2 object

Using map() method

By using map() method to create key/value pairs, Pair RDD can be created

scala> val inputrdd   = sc.parallelize(List("key1 1", "key1 3"))
inputrdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[15] at parallelize at :21

scala> val mapped     = inputrdd.map { x =>
     |    val splitted = x.split(" ")
     |    (splitted(0), splitted(1).toInt)
     | }
mapped: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[16] at map at :23

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

scala> reduced.collect()
res24: Array[(String, Int)] = Array((key1,4))

A Word Count Application

In this example, we will count the number of occurrence of each unique word in a given file

object WordCount {
   def main(args: Array[String]) {

      println("In main : " + args(0) + "," + args(1))

      //Create Spark Context
      val conf = new SparkConf().setAppName("WordCountApp")
      val sc = new SparkContext(conf)

      //Load Data from File
      val input = sc.textFile(args(0))

      //Split into words
      val words = input.flatMap(line => line.split(" "))

      //Assign unit to each word
      val units = words.map ( word => (word, 1) )

      //Reduce each key
      val counts = units.reduceByKey ( (x, y) => x + y )

      //Write output to Disk
      counts.saveAsTextFile(args(1))

      //Shutdown spark. System.exit(0) or sys.exit() can also be used
      sc.stop();
   }
}

'object'

object WordCount {

In Scala we use the keyword object to create a singleton. A class can be created using keyword class

'val' & Type Inference

def main(args: Array[String]) { 

A method in Scala is defined using the Keyword def.  A function defined inside class/object is called as Method

val sc = new SparkContext(conf)

We are creating a Spark context to access Cluster. In Scala,  The keyword val specifies Immutability, which actually means we cannot change the value of sc once defined. We can also create a value using the keyword var which specifies mutability. Using immutable value(val) is the preferred approach in Scala

A val or var is always associated with a type. If we do not specify the type explicitly(as in this case), Scala compiler will infer the type based on the data. This is called as Type Inference

Higher Order Functions & Function Literals

val words = input.flatMap(line => line.split(" "))

In Scala, an argument to the method can either be var, val (or) a function. A function that takes another function as argument is called Higher Order Functions. Here the method flatMap is a higher order function and we are passing an anonymous function(A function without name) as argument. In Scala, anonymous functions are called as Function Literals. In Java, anonymous functions are called as lambdas. Checkout this example to know more about flatMap

Implicit Conversion

val units = words.map ( word => (word, 1) )
val counts = units.reduceByKey ( (x, y) => x + y )
Here units is of type RDD. As can be seen in the API document, the type RDD do not have the method reduceByKey. Why did not the Scala compiler throw any error? The reason is Implicit Conversion. When the compiler finds an unknown field or method in the current instance(units), it converts the current type(in our case RDD) into Another type which has this method(reduceByKey) defined. In our case the compiler converts the type RDD to type PairRDDFunctions (which has the method reduceByKey defined) so that reduceByKey can be called

In order for this Implicit conversion to work, we would need to import  org.apache.spark.SparkContext._

Here is the input data we are going to test

mountain@mountain:~/sbook$ cat src/main/resources/data.txt 
line1 word1 word2
line2 word3
line3 word1

Let us compile our code & package it into jar

mountain@mountain:~/sbook$ sbt package
...
[info] Packaging /home/mountain/workspace/allgitrepositories/ApacheSparkBook/target/scala-2.11/sparkbookapp_2.11-1.0.jar ...

Submit the Application to Spark

spark-submit --class sbook.helloworld.WordCount target/scala-2.11/sparkbookapp_2.11-1.0.jar src/main/resources/data.txt  output

Now we have the result available in directory output

mountain@mountain:~/sbook$ ls output

part-00000  part-00001  _SUCCESS

mountain@mountain:~/sbook$ cat output/*

(word2,1)

(line2,1)

(word3,1)

(line3,1)

(line1,1)

(word1,2)

That ends our first Scala Application in Spark