Serialization

Most of the Transformation & some actions are Higher order functions, so they take another Function as an Argument

Ex : val filteredRDD = inputRDD.filter(x => x != 5)

Here the transformation filter() is taking another function(in this case Function Literal / Anonymous function) as an argument. As this function argument gets executed in different Executors, it is imperative, these function argument are serializable, so that they can be transferred to the Executors. 

If the function we pass as an argument is non-serializable, then NotSerializableException occurs

Member Variable & Methods

Passing a Member Variable (or) Method of an Object to a Transformation (or) Action results in a complete reference to the object, because of which Spark will transfer the Complete object to the Executors in which it(The Member variable or Method) will be referenced. This results in unnecessary transfer of data(ie.. in order to access one variable/method we end up transferring the entire object). In order to avoid this, always copy the Member variable/method into a local variable before passing it as an argument to the Transformation (or) Action

class Append(val suffix: String) {
  
  def appendSuffixBad(inputRDD: RDD[String]): RDD[String] = {
     // Bad      As we are referring the Member variable 'suffix'
     //          the whole object will be serialized & transferred
     //          to the Executor(so that 'suffix' can be referred)
     inputRDD.map { x => x + suffix }             
  }
  
  def appendSuffixGood(inputRDD: RDD[String]): RDD[String] = {
     // Good      Here we copied the Member variable 'suffix' into a 
     //           local variable, so only the local variable is transferred
     //           to the executor
     val localsuffix = suffix;
     inputRDD.map { x => x + localsuffix }             
  }

}

Reference

Learning Spark : Page 32

RDD Persistence

RDDs are Recomputed

RDDs by default is recomputed each time an action is run on them. For example,

scala> val lines = sc.textFile("words.txt")
...
scala> lines.first()
res4: String = line1 word1
scala> lines.count()
res5: Long = 4

Here the call to action first() computes the RDD 'lines'. Again when we use another action 'count()' on the same RDD, the RDD is recomputed once again

Persisting RDDs

The default behavior of recomputing the RDDs on each action can be overridden by persisting the RDDs, so that no re-computation is done each time an action is called on the RDD. When persisted, each node that compute the RDD store the result in their Partitions

We use persist() method to persist an RDD. In Scala & Java, by default, persist() will store the data in JVM as unserialized object. In Python, calling persist() will serialize the data before persisting. Options to store in Memory/Disk combination is also possible.

scala> val lines = sc.textFile("words.txt")

...

scala> import org.apache.spark.storage.StorageLevel

...

scala> lines.persist(StorageLevel.MEMORY_ONLY) //We can also use cache() method if we need MEMORY_ONLY storage level

...

scala> lines.count() (1)

...

The actual persistence takes place during the first (1) action call on the RDD. Spark provides multiple Storage options(Memory/Disk) to persist the data as well as Replication Levels. More information can be found here

We use unpersist() to unpersist RDD. When the cached data exceeds the Memory capacity, Spark automatically evicts the old partitions(it will be recalculated when needed). This is called Last Recently used Cache(LRU) policy