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

Line Count using Scala Shell

In this example, we will count the number of lines in a Text File. Create a text file with random Content
mountain@mountain:~$ cat data.txt
line 1
line 2

Open Scala Shell and execute the following Statements

mountain@mountain:~$ spark-shell

...
Spark context available as sc (1)
scala> val lines = sc.textFile("data.txt") (2)

...

lines: org.apache.spark.rdd.RDD[String] = data.txt MapPartitionsRDD[1] at textFile at <console>:21

scala> lines.count() (3)

...

res0: Long = 2

scala> lines.first()

...

res1: String = line 1

You can exit from shell using,
scala> exit

Controlling the Logs

You must have realized by now, the Scala Shell prints overwhelming amount of logs which you might not need all the time. The amount of logs printed on the Shell can be controlled by setting the property log4j.rootCategory in file log4j.properties

In $SPARK_HOME/conf,
mountain@mountain:~/sk/conf$ cp log4j.properties.template log4j.properties
mountain@mountain:~/sk/conf$ vi log4j.properties
log4j.rootCategory=WARN, console

Setting the log level to WARN will reduce the amount of logs printed on the Shell. Let us dissect the output from our simple program and grasp few important concepts of Spark

Driver

A Driver Program is akin to main() method in programming languages like C, C++ & java. The Driver Program which is part of a Spark Application launches the Application into Spark Cluster. In our case, The Scala Shell acts as a Driver Program

Spark Context (sc)

In Spark, we access the cluster through object of type SparkContext. The Scala shell by default provides the variable sc(1) which is of type SparkContext 

Resilient Distributed Datasets(RDD)

RDD is the basic unit of Data in Spark upon which all Operations are done. In the example above, lines(2) is an RDD. count()(3) is one of the Operation we have performed on the RDD

An RDD Operation can be any one of this type, actions or transformations.

action returns result to the Driver Program or write it to the Storage. An action normally starts a Computation to provide result and always return some other data type other than RDD

transformation returns Pointer to new RDD

Check the link below for some of the common actions & transformations

https://spark.apache.org/docs/latest/programming-guide.html#actions
https://spark.apache.org/docs/latest/programming-guide.html#transformations

In a typical cluster environment an operation is normally parallelized in multiple Nodes.