Understanding Spark serialization

In Spark how does one know which objects are instantiated on driver and which are instantiated on executor , and hence how does one determine which classes needs to implement Serializable ?

Solution 1:

To serialize an object means to convert its state to a byte stream so that the byte stream can be reverted back into a copy of the object. A Java object is serializable if its class or any of its superclasses implements either the java.io.Serializable interface or its subinterface, java.io.Externalizable.

A class is never serialized only object of a class is serialized. Object serialization is needed if object needs to be persisted or transmitted over the network .

Class Component            Serialization
instance variable           yes
Static instance variable    no
methods                     no
Static methods              no
Static inner class          no
local variables             no

Let's take a sample Spark code and go through various scenarios

public class SparkSample {

      public int instanceVariable                =10 ;
      public static int staticInstanceVariable   =20 ;

      public int run(){

         int localVariable                       =30;

         // create Spark conf
         final SparkConf sparkConf = new SparkConf().setAppName(config.get(JOB_NAME).set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");

         // create spark context 
         final JavaSparkContext sparkContext = new JavaSparkContext(sparkConf);

        // read DATA 
        JavaRDD<String> lines = spark.read().textFile(args[0]).javaRDD(); 


        // Anonymous class used for lambda implementation
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
                @Override
                public Iterator<String> call(String s) {
                // How will the listed varibles be accessed in RDD across driver and Executors 
                System.out.println("Output :" + instanceVariable + " " + staticInstanceVariable + " " + localVariable);
                return Arrays.asList(SPACE.split(s)).iterator();
        });

        // SAVE OUTPUT
        words.saveAsTextFile(OUTPUT_PATH));

      }

       // Inner Static class for the funactional interface which can replace the lambda implementation above 
       public static class MapClass extends FlatMapFunction<String, String>() {
                @Override
                public Iterator<String> call(String s) {
                System.out.println("Output :" + instanceVariable + " " + staticInstanceVariable + " " + localVariable);
                return Arrays.asList(SPACE.split(s)).iterator();
        }); 

        public static void main(String[] args) throws Exception {
            JavaWordCount count = new JavaWordCount();
            count.run();
        }
}

Accessibility and Serializability of instance variable from Outer Class inside inner class objects

    Inner class        | Instance Variable (Outer class) | Static Instance Variable (Outer class) | Local Variable (Outer class)

    Anonymous class    | Accessible And Serialized       | Accessible yet not Serialized          | Accessible And Serialized 

    Inner Static class | Not Accessible                  | Accessible yet not Serialized          | Not Accessible 

Rule of thumb while understanding Spark job is :

1. All the lambda functions written inside the RDD are instantiated on the driver and the objects are serialized and sent to the executors

2. If any outer class instance variables are accessed within the inner class, compiler apply different logic to access them, hence outer class gets serialized or not depends what do you access.

3. In terms of Java, the whole debate is about Outer class vs Inner class and how does accessing outer class references and variables leads to serialization issues.

Various scenarios:

Outer class Variable variables accessed within Anonymous class :

Instance Variable (Outer class)

Compiler by default inserts constructor in the byte code of the

Anonymous class with reference to Outer class object .

The outer class object is used to access the instance variable

Anonymous-class(){

 final Outer-class reference;

 Anonymous-class( Outer-class outer-reference){

reference = outer-reference;

}

}

The outer class is serialized and sent along with the serialized object of the inner anonymous class

Static Instance Variable (Outer class)

As static variables are not serialized , outer class object is still inserted into the Anonymous class constructor .

The value of the static variable is taken from the class state

present on that executor .

Local Variable (Outer class)

Compiler by default inserts constructor in the byte code of the

Anonymous class with reference to Outer class object AND local variable refrence.

The outer class object is used to access the instance variable

Anonymous-class(){

 final Outer-class reference;

final Local-variable localRefrence ;

 Anonymous-class( Outer-class outer-reference, Local-variable localRefrence){

reference = outer-reference;

this.localRefrence = localRefrence;

}

}

The outer class is serialized , and the local variable object is also

serialized and sent along with the serialized object of the inner anonymous class

As the local variable becomes a instance member inside the anonymous class it needs to be serialized . From outer class perspective the local variable can never be serialized

----------

Outer class variables accessed with Static inner class.

Instance Variable (Outer class)

cant be accessed

Local Variable (Outer class)

cant be accessed

Static Instance Variable (Outer class)

As static variables are not serialized hence no outer class object is serialized.

The value of the static variable is taken from the class state

present on that executor .

Outer class is not serialized and send along with the serialized Static inner class

Points to think through:

1. Java Serialization rules are followed to select which class object needs to be serialized .

2. Use javap -p -c "abc.class" to unwrap the byte code and see the compiler generated code

3. Depending on what you are trying to access within the inner class of the outer class, compiler generates different byte code.

4. You don't need to make classes implement Serialization which are only accessed on driver .

5. Any anonymous/static class (all lambda function are anonymous class) used within RDD will be instantiated on the driver .

6. Any class/variable used inside RDD will be instantiated on driver and sent to the executors .

7. Any instance variable declared transient will not be serialized on driver.

   8. By default Anonymous classes will force you to make the outer class serializable.
   9. Any local variable/object need not have to be serializable .
   10. Only if local variable is used inside the Anonymous class needs to be serialized
   11. One can create singleton inside the call() method of pair,mapToPair function , thus making sure its never initialized on driver
   12. static variables are never serialized hence are never sent from driver to executors
8. if u need any service to be executed only on the executor , make them static fields inside the lambda function , or make them transient and singelton and check for null condition to instantiate them

Solution 2:

there are plenty of very well written blogs that explained this very well, like this one: spark serialization challenges.

but in short, we can conclude like this(Spark only, not JVM in general):

1. due to JVM, only objects can be serialized(functions are objects)
2. if an object needs to be serialized, it's parent object needs to be serialized as well
3. any Spark operations like(map, flatMap, filter, foreachPartition, mapPartition etc..), if inner part has reference to outer part object, that object needs to be serialized. Because outer part objects are in Driver, not in Executors. And serialization policy refers to my point #2.
4. reference to Scala object(aka, Scala singleton) will not get serialized(for mapPartition and foreachPartition only, UDF will always get serde from driver to executors). executors will directly refer to their local JVM's object, since it's a singleton it will exist on executors JVM. This means, Driver's mutation on its local object won't be seen from executors.