apache-spark - started - apache spark windows

Requiere serialización de kryo en Spark(Scala)

Question

Tengo la serialización de kryo activada con esto:

conf.set( "spark.serializer", "org.apache.spark.serializer.KryoSerializer" )

Quiero asegurarme de que una clase personalizada se serialice usando kryo cuando se mezcla entre nodos. Puedo registrar la clase con kryo de esta manera:

conf.registerKryoClasses(Array(classOf[Foo]))

Tal como lo entiendo, esto en realidad no garantiza que se utilice la serialización kyro; si un serializador no está disponible, Kryo recurrirá a la serialización de Java.

Para garantizar que la serialización de kryo ocurra, seguí esta recomendación de la documentación de Spark:

conf.set("spark.kryo.registrationRequired", "true")

Pero esto hace que IllegalArugmentException sea lanzado ("La clase no está registrada") para un grupo de diferentes clases que supongo que Spark usa internamente, por ejemplo, lo siguiente:

org.apache.spark.util.collection.CompactBuffer
scala.Tuple3

Seguramente no tengo que registrar manualmente cada una de estas clases individuales con kryo? Estos serializadores están todos definidos en kryo, entonces ¿hay alguna forma de registrarlos automáticamente?

Answer 1

Tal como lo entiendo, esto en realidad no garantiza que se utilice la serialización kyro; si un serializador no está disponible, Kryo recurrirá a la serialización de Java.

No. Si configura spark.serializer en org.apache.spark.serializer. KryoSerializer org.apache.spark.serializer. KryoSerializer luego Spark usará Kryo. Si Kryo no está disponible, obtendrá un error. No hay retroceso.

Entonces, ¿qué es este registro de Kryo?

Cuando Kryo serializa una instancia de una clase no registrada, tiene que generar el nombre de clase completamente calificado. Eso es un montón de personajes. En cambio, si una clase ha sido prerregistrada, Kryo solo puede generar una referencia numérica a esta clase, que tiene solo 1-2 bytes.

Esto es especialmente crucial cuando cada fila de un RDD se serializa con Kryo. No desea incluir el mismo nombre de clase para cada mil millones de filas. Entonces preinscribe estas clases. Pero es fácil olvidarse de registrar una nueva clase y luego perderá bytes nuevamente. La solución es exigir que todas las clases estén registradas:

conf.set("spark.kryo.registrationRequired", "true")

Ahora Kryo nunca dará salida a nombres completos de clase. Si encuentra una clase no registrada, es un error de tiempo de ejecución.

Desafortunadamente es difícil enumerar todas las clases que vas a serializar de antemano. La idea es que Spark registre las clases específicas de Spark, y registrar todo lo demás. Tienes un RDD[(X, Y, Z)] ? Debes registrar classOf[scala.Tuple3[_, _, _]] .

La lista de clases que Spark registra en realidad incluye CompactBuffer , por lo que si ve un error al respecto, está haciendo algo mal. Está pasando por alto el procedimiento de registro Spark. Debes usar spark.kryo.classesToRegister o spark.kryo.registrator para registrar tus clases. (Consulte las opciones de configuración . Si usa GraphX, su registrador debe llamar a GraphXUtils. RegisterKryoClasses ).

Answer 2

Según lo que estás viendo, lo mejor es que te falta la declaración:

sparkConf.set( "spark.serializer", "org.apache.spark.serializer.KryoSerializer" )

En los últimos días también he estado luchando con la conversión de la serialización a Kryo, incluso para GraphX, incluido el registro de scala.Tuple3 con Kryo, aparentemente porque el código Spark / GraphX ​​está creando un Tuple3 cuando hago un ''sortBy''.

He estado agregando un montón de otras clases, una por una, a la lista para registrarme en Kryo, principalmente clases de Scala y Spark que no creo que tenga que agregar. Pensando / esperando que haya una mejor manera de usar Kryo con Spark.