serialization - example - Comprender la serialización de Spark

spark sql example

Question

En Spark, ¿cómo se sabe qué objetos se instancian en el controlador y cuáles se ejecutan en el ejecutor, y por lo tanto, cómo se determina qué clases necesita implementar Serializable?

Answer 1

Hay muchos blogs muy bien escritos que explican esto muy bien, como este: provocar desafíos de serialización .

pero en resumen, podemos concluir así (solo Spark, no JVM en general):

1. debido a JVM, solo los objetos pueden ser serializados (las funciones son objetos)
2. si un objeto necesita ser serializado, su objeto padre también debe ser serializado
3. cualquier operación de Spark como (map, flatMap, filter, foreachPartition, mapPartition, etc.), si la parte interna tiene referencia al objeto de la parte externa, ese objeto debe ser serializado. Debido a que los objetos de la parte externa están en el controlador, no en los ejecutores. Y la política de serialización se refiere a mi punto # 2.
4. la referencia al object Scala (también conocido como Scala singleton) no se serializará. los ejecutores se referirán directamente al objeto de su JVM local, ya que es un singleton que existirá en los JVM de los ejecutores. Esto significa que la mutación del conductor en su object local no será vista por los ejecutores.

Answer 2

Serializar un objeto significa convertir su estado en una secuencia de bytes para que la secuencia de bytes se pueda revertir en una copia del objeto. Un objeto Java es serializable si su clase o cualquiera de sus superclases implementa la interfaz java.io.Serializable o su subinterfaz, java.io.Externalizable.

Una clase nunca se serializa, solo el objeto de una clase se serializa. La serialización de objetos es necesaria si el objeto necesita persistirse o transmitirse a través de la red.

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

Tomemos un código Spark de muestra y veamos varios escenarios

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();
}
}

Accesibilidad y serialización de la variable de instancia de la clase externa dentro de los objetos de la clase interna

 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

La regla general al comprender el trabajo de Spark es:

1. Todas las funciones lambda escritas dentro del RDD se instancian en el controlador y los objetos se serializan y se envían a los ejecutores

2. Si se accede a alguna variable de instancia de clase externa dentro de la clase interna, el compilador aplica una lógica diferente para acceder a ellas, por lo tanto, la clase externa se serializa o no depende de a qué se acceda.

3. En términos de Java, todo el debate es sobre la clase externa vs clase interna y cómo el acceso a las referencias y variables de la clase externa conduce a problemas de serialización.

Varios escenarios:

Clase externa Variables variables a las que se accede dentro de la clase Anónimo:

Variable de instancia (clase externa)

El compilador por defecto inserta el constructor en el código de bytes del

Clase anónima con referencia al objeto de clase externa.

El objeto de clase externa se usa para acceder a la variable de instancia

Anonymous-class () {

 final Outer-class reference;
Anonymous-class( Outer-class outer-reference){
reference = outer-reference;
}

}

La clase externa se serializa y se envía junto con el objeto serializado de la clase anónima interna.

Variable de instancia estática (clase externa)

Como las variables estáticas no se serializan, el objeto de clase externa todavía se inserta en el constructor de la clase Anónimo.

El valor de la variable estática se toma del estado de clase

presente en ese ejecutor.

Variable local (clase externa)

El compilador por defecto inserta el constructor en el código de bytes del

Clase anónima con referencia al objeto de clase externa Y referencia de variable local.

El objeto de clase externa se usa para acceder a la variable de instancia

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;
}

}

La clase externa está serializada y el objeto variable local también

serializado y enviado junto con el objeto serializado de la clase anónima interna

A medida que la variable local se convierte en un miembro de instancia dentro de la clase anónima, debe ser serializada. Desde la perspectiva de la clase externa, la variable local nunca se puede serializar

----------

Variables de clase externa accedidas con clase interna estática.

Variable de instancia (clase externa)

no se puede acceder

Variable local (clase externa)

no se puede acceder

Variable de instancia estática (clase externa)

Como las variables estáticas no se serializan, por lo tanto, no se serializa ningún objeto de clase externa.

El valor de la variable estática se toma del estado de clase

presente en ese ejecutor.

La clase externa no está serializada y se envía junto con la clase interna estática serializada

Puntos para pensar:

1. Se siguen las reglas de serialización de Java para seleccionar qué objeto de clase se debe serializar.

2. Use javap -p -c "abc.class" para desenvolver el código de bytes y ver el código generado por el compilador

3. Dependiendo de lo que intente acceder dentro de la clase interna de la clase externa, el compilador genera un código de bytes diferente.

4. No es necesario que las clases implementen la serialización a la que solo se accede desde el controlador.

5. Cualquier clase anónima / estática (todas las funciones lambda son de clase anónima) utilizada dentro de RDD se instanciará en el controlador.

6. Cualquier clase / variable utilizada dentro de RDD se instanciará en el controlador y se enviará a los ejecutores.

7. Cualquier variable de instancia declarada transitoria no se serializará en el controlador.

   8. Por defecto, las clases anónimas te obligarán a hacer que la clase externa sea serializable.
   9. Cualquier variable / objeto local no necesita ser serializable.
   10. Solo si la variable local se usa dentro de la clase Anónimo necesita ser serializada
   11. Se puede crear singleton dentro del método de par call (), función mapToPair, asegurándose de que nunca se inicialice en el controlador
   12. las variables estáticas nunca se serializan, por lo tanto, nunca se envían desde el controlador a los ejecutores
8. si necesita que algún servicio se ejecute solo en el ejecutor, conviértalos en campos estáticos dentro de la función lambda, o hágalos transitorios y singelton y compruebe la condición nula para instanciarlos