Spark Fundamentals - RDD

The main abstraction Spark provides is a resilient distributed dataset (RDD)

Resilient Distributed Datasets (RDD) is a fundamental data structure of Spark. It is an immutable distributed collection of objects. Each dataset in RDD is divided into logical partitions, which may be computed on different nodes of the cluster.

RDD is a logical reference of a dataset which is partitioned across many server machines in the cluster. RDDs are Immutable and are self recovered in case of failure.

dataset could be the data loaded externally by the user. It could be a json file, csv file or a text file with no specific data structure.

RDDs have the following properties –

1. Immutability and partitioning: RDDs composed of collection of records which are partitioned. Partition is basic unit of parallelism in a RDD, and each partition is one logical division of data which is immutable and created through some transformations on existing partitions.Immutability helps to achieve consistency in computations.
   Users can define their own criteria for partitioning based on keys on which they want to join multiple datasets if needed.
2. Coarse grained operations: Coarse grained operations are operations which are applied to all elements in datasets. For example – a map, or filter or groupBy operation which will be performed on all elements in a partition of RDD.
3. Fault Tolerance: Since RDDs are created over a set of transformations , it logs those transformations, rather than actual data.Graph of these transformations to produce one RDD is called as Lineage Graph.

For example –

firstRDD=sc.textFile("hdfs://...")

secondRDD=firstRDD.filter(someFunction);

thirdRDD = secondRDD.map(someFunction);

result = thirdRDD.count()

In case of we lose some partition of RDD , we can replay the transformation on that partition in lineage to achieve the same computation, rather than doing data replication across multiple nodes.This characteristic is biggest benefit of RDD , because it saves a lot of efforts in data management and replication and thus achieves faster computations.

4. Lazy evaluations: Spark computes RDDs lazily the first time they are used in an action, so that it can pipeline transformations. So , in above example RDD will be evaluated only when count() action is invoked.
5. Persistence: Users can indicate which RDDs they will reuse and choose a storage strategy for them (e.g., in-memory storage or on Disk etc.)

These properties of RDDs make them useful for fast computations.

Setting up Spark-scala development environment

Pre-requisites:

1. Oracle Virtualbox
    Download and install from https://www.virtualbox.org/ and create a ubuntu VM

2. Java1.7 or 1.8

3. Scala 2.x.x (any version)
    Download and install scala. Setup home and PATH
    https://www.scala-lang.org/download/install.html

4. Install sbt from http://www.scala-sbt.org/0.13/docs/Using-sbt.html
    Setup home and path

5. Install and setup spark from
    http://creativelytechie.blogspot.ca/2016/07/steps-to-installing-spark.html
    Ensure version compatibility of spark with scala


5. Eclipse IDE in VM
    Install eclipse IDE (new or upgrade existing eclipse for scala) from 
    http://scala-ide.org/download/prev-stable.html

Learning Scala - Session 5 - Case Class & Singleton

Case class in Scala

- Creates an object without using new

package com.first

case class Person(lastname:String, firstname: String, age:Int) {
  println(s"Person is $firstname $lastname, age is $age")
}

object PersonObj extends App{
  val person1 = Person("Shree","Anita",30)
  val person2 = Person("Kumar", "Saahil", 30)
  val person3 = person1.copy(firstname="Shilpa",age=29)
}

Output
Person is Anita Shree, age is 30

Person is Saahil Kumar, age is 30
Person is Shilpa Shree, age is 29 (output)

When you declare a case class the Scala compiler does the following for you:

• Creates a class and its companion object.
• Implements the apply method that you can use as a factory. This lets you create instances of the class without the new keyword. E.g.:
• Prefixes all arguments, in the parameter list, with val. This means the class is immutable, hence you get the accessors but no mutators. 
• Adds natural implementations of hashCode, equals and toString. Since == in Scala always delegates to equals, this means that case class instances are always compared structurally
• Generates a copy method to your class to create other instances starting from another one and keeping some arguments the same. E.g.: Create another instance keeping the lastname and changing firstname and age:
• val person3 = person1.copy(firstname="Manu",age=29)
• Person is Manu Shree, age is 29 (output)
• Please refer http://www.alessandrolacava.com/blog/scala-case-classes-in-depth/ for in depth details on case class

Implementing Singleton in Scala

When you replace the keyword class by object, you define a singleton object. A singleton definition defines a new type like a class does, but only one single object of this type can be created—therefore the name "singleton".

Singleton.scala

object Singleton {

  println("Creating Singleton Object")

  

  var sum = 0;

  def add(num:Int){

    sum += num

      println(s"Sum =$sum")

  }

}

Main.scala

object Main extends App{

  Singleton.add(10)

  Singleton.add(5)

  Singleton.add(20)

}

Output

Creating Singleton Object

Sum =10

Sum =15

Sum =35

The definition will create an object of this type, and assign the name MySingleton to it. The construction of the object happens the first time the object is used (so if you define a singleton and never use it, it will not be constructed at all).

You can define both a class and a singleton object with the same name. In this case, the singleton is called the companion object of the class. A companion object is allowed to use the private fields and private methods of a class.

Learning Scala - Session 4(cont) - Constructors

Constructors in Scala

2 types - Primary and Auxiliary

Primary Constructor

//Primary constructor to accept 2 integer variable and compute the sum
class ConstructorTest(val num1:Int,val num2:Int){
   var sum = num1+num2
   println(s"Sum of $num1 and $num2 = $sum")
}

- It is part of the class body itself
- Variables are defined with the class body
- All statements within the class are considered to be part of the primary constructor
- Compiler converts the variables to fields and calls getter method for val (Immutable) and setter and getter for var (Mutable)

The same class in java would be
class ConstructorTest{
         public final int num1;
         public final int num2;
       
         public void ConstructorTest(int num1, int num2){
                   this.num1 = num1;
                   this.num2 = num2;
         }

         public int sum(int num1, int num2){
                    int sum = num1+num2;
                    println("Sum of " + num1 + " + " num2 + " = " + sum);
         }
}

So many lines of code in Java, is implemented in just 3 lines of code in Scala using Primary Constructor


Auxiliary Constructor
These are same as Java except that
- always named as 'this'
- The first line of code within an auxiliary constructor has to call either another auxiliary constructor or the primary one. This ensures that the object is going through the whole process of initialization.
- Examples below

//Auxiliary constructor1 to accept 3 integer variable and compute the sum of 2 integers using primary constructor and multiply the result by 3^rd Integer

def this(num1:Int, num2:Int, num3:Int){
    this(num1,num2) //calls the primary constructor
    product = num3*sum;
    println(s"Product of ($num1 and $num2)*$num3 = $product")
}

//Auxiliary constructor2 to accept 4 integer variable and compute the result of 3 integers using auxiliary constructor1 and divide the result by 4th Integer

def this(num1:Int, num2:Int, num3:Int, num4:Int){
    this(num1,num2,num3)
    division = product/num4
    println(s"Division of $product/$num4 = $division")
  }

Object creation for the above class would simply be

object ConstructorObj extends App{
     var obj = new ConstructorTest(10,5,2,3)
     println(obj)
}


Output of the executing the ConstrutorObj is

Constructor Test
***********************
Sum of 10 and 5 = 15
Product of (10 and 5)*2 = 30
Division of 30/3 = 10
Result: sum=15, product=30, division=10

Complete code for a sample demo of Primary and Auxiliary constructors can be found here

Learning Scala - Session 4 - First Class

First.scala class 

package com.first

class First {
  var name = ""
  var age = 0
  override def toString=s"name is $name, age is $age";  //override toString helps to directly print the obj

  object FirstObj {
    var obj = new First()
    obj.name = "Tanu"
    obj.age = 30
   
    def main(args:Array[String]){
      println(obj)   //directly print object if toString is overriden, if not, then prints reference
    }
  }
}

Output:
name is Tanu, age is 30

To run a scala program, we need to run the object and not the class

obj extends App

 object FirstObj extends App{
    var obj = new First()
    obj.name = "Tanu"
    obj.age = 30
      println(obj)
  }

extends App - by default puts all to main method, no need to define the main method. App is a Trait

Trait in scala is same as Interface with default implementation (more on this later)

Learning Scala - Session 3 - String operations

String Functions

drop(int) - drop drops the characters mentioned as arguments.
Can be used with capitalize function, which makes only the first letter capital.
Highly needed to be called directly in Big Data, speeds up string processing

scala> var x = "1: test is a string"                                                                                                                                            

x: String = 1: test is a string                                                                                                                                                 

                                                                                                                                                                                

scala> x.drop(2).capitalize                                                                                                                                                     

res1: String = " test is a string"

take() - takes the number of characters mentioned in the argument

scala> var x = "row1:test is a string"                                                                                                                                          

x: String = row1:test is a string

scala> x.take(4)                                                                                                                                                                

res1: String = row1 

split() - splits the string with the given string into a string Array

scala> var x = "test is a string"                                                                                                                                               

x: String = test is a string 

scala> x.split(" ")                                                                                                                                                             

res2: Array[String] = Array(test, is, a, string)

scala> x.split("is")                                                                                                                                                            

res0: Array[String] = Array("test ", " a string") 

getBytes - gives the ASCII values of chars

scala> x.getBytes                                                                                                                                                               

res1: Array[Byte] = Array(116, 101, 115, 116, 32, 105, 115, 32, 97, 32, 115, 116, 114, 105, 110, 103) 

x.<tab> - will give available functions in the base class 

String comparisons

== operator is used for comparing two strings and returns true or false

scala> var x ="test is a string"                                                                                                                                                

x: String = test is a string 

scala> var y = "hello"                                                                                                                                                          

y: String = hello                                                                                                                                                               

                                                                                                                                                                                

scala> x==y                                                                                                                                                                     

res2: Boolean = false    

Comparing with null does not give null pointer exception, because first it checks size of both strings, If sizes are different, it returns false

scala> y=null                                                                                                                                                                   

y: String = null                                                                                                                                                                

                                                                                                                                                                                

scala> x==y                                                                                                                                                                     

res3: Boolean = false

On the other hand, if x.trim()==y.trim() gives Null pointer exception, because it first calls the function. null.trim() gives Null pointer exception

scala> x.trim()==y.trim()                                                                                                                                                       

java.lang.NullPointerException 

Println and variable assignments for strings

raw
println("Hello \n world") - prints hello, newline, world
In Scala, raw - means print as it is
println(raw"Hello \n world")

scala> println("Hello \n world")                                                                                                                                                

Hello                                                                                                                                                                           

 world                                                                                                                                                                          

                                                                                                                                                                                

scala> println(raw"Hello \n world")                                                                                                                                             

Hello \n world 

scala> var str =raw"Hello \n world"                                                                                                                                             

str: String = Hello \n world                                                                                                                                                    

                                                                                                                                                                                

scala> println(str)                                                                                                                                                             

Hello \n world 

s-interpolator (Whenever a statement begins with s is understood by the compiler to replace $ with actual values)
To print values in java, we need + "" etc
In Scala, with s or call to Interpolator, its much easier, as below

scala> var name = "tanushree"                                                                                                                                                   

name: String = tanushree                                                                                                                                                                                                                                                                                                                                  

scala> var comp="igate"                                                                                                                                                         

comp: String = igate                                                                                                                                                            

In Java                                                                                                                                                                                

scala> println("name:"+name+"works in company:"+comp)                                                                                                                           

name:tanushreeworks in company:igate                                                                                                                                            

       

In Scala                                                                                                                                                                         

scala> println(s"name:$name works in company:$comp")                                                                                                                            

name:tanushree works in company:igate           

Call to toUpperCase
"Hello World".map(c=>c.toUpper) is same as
"Hello World".toUpperCase (internally calls map method)
In big data processing, directly calling map avoids one extra method call and improves processing

scala> "Hello World".map(c=>c.toUpper)                                                                                                                                          

res2: String = HELLO WORLD                                                                                                                                                      

                                                                                                                                                                                

scala> "Hello World".toUpperCase                                                                                                                                                

res3: String = HELLO WORLD  

TypeCasting from Str to int, long

scala> var numStr ="1234"                                                                                                                                                       

numStr: String = 1234                                                                                                                                                           

                                                                                                                                                                                

scala> numStr.toInt                                                                                                                                                             

res2: Int = 1234                                                                                                                                                                

                                                                                                                                                                                

scala> numStr.toLong                                                                                                                                                            

res3: Long = 1234