Tarea no serializable: java. io. NotSerializableException al llamar a la función fuera del cierre solo en clases no objetos

La respuesta de Grega es excelente para explicar por qué el código original no funciona y dos formas de solucionar el problema. Sin embargo, esta solución no es muy flexible; considere el caso en el que su cierre incluye una llamada a un método en una clase no-Serializable sobre la que no tiene control. No puede agregar la etiqueta Serializable a esta clase ni cambiar la implementación subyacente para cambiar el método en una función.

Nilesh presenta una gran solución para esto, pero la solución puede ser hecho más conciso y general:

def genMapper[A, B](f: A => B): A => B = {
  val locker = com.twitter.chill.MeatLocker(f)
  x => locker.get.apply(x)
}

Este serializador de funciones se puede usar para envolver automáticamente cierres y llamadas a métodos:

rdd map genMapper(someFunc)

Esta técnica también tiene el beneficio de no requerir las dependencias adicionales de Shark para acceder KryoSerializationWrapper, ya que el Chill de Twitter ya es atraído por core Spark

Charla completa explicando completamente el problema, que propone un gran cambio de paradigma para evitar estos problemas de serialización: https://github.com/samthebest/dump/blob/master/sams-scala-tutorial/serialization-exceptions-and-memory-leaks-no-ws.md

La respuesta más votada básicamente sugiere desechar una característica completa del lenguaje, que ya no es usar métodos y solo usar funciones. De hecho, en los métodos de programación funcional en las clases debe evitarse, pero convertirlos en funciones no está resolviendo el problema de diseño aquí (ver el enlace anterior).

Como solución rápida en esta situación en particular, podría usar la anotación @transient para decirle que no intente serializar el valor ofensivo (aquí, Spark.ctx es una clase personalizada, no la de Spark después del nombre de OP):

@transient
val rddList = Spark.ctx.parallelize(list)

También puede reestructurar el código para que rddList viva en otro lugar, pero eso también es desagradable.

El Futuro es Probablemente Esporas

En el futuro Scala incluirá estas cosas llamadas "esporas" que deberían permitirnos controlar el grano fino lo que hace y no exactamente es arrastrado por un cierre. Además, esto debería convertir todos los errores de extraer accidentalmente tipos no serializables (o cualquier valor no deseado) en errores de compilación en lugar de ahora, que es horrible excepciones de tiempo de ejecución / fugas de memoria.

Http://docs.scala-lang.org/sips/pending/spores.html

Un consejo sobre la serialización de Kryo

Cuando use kyro, hágalo de manera que el registro es necesario, esto significará que obtendrá errores en lugar de fugas de memoria:

"Finalmente, sé que kryo tiene kryo.setRegistrationOptional (verdadero) pero estoy teniendo un tiempo muy difícil tratando de averiguar cómo usarlo. Cuando esta opción está activada, kryo todavía parece lanzar excepciones si no he registrado clases."

Estrategia para registrar clases con kryo

Por supuesto, esto solo le da control a nivel de tipo, no a nivel de valor control.

... más ideas por venir.

Resolví este problema usando un enfoque diferente. Simplemente necesita serializar los objetos antes de pasar por el cierre, y des-serializar después. Este enfoque simplemente funciona, incluso si sus clases no son serializables, porque utiliza Kryo detrás de las escenas. Todo lo que necesitas es curry. ;)

Aquí hay un ejemplo de cómo lo hice:

def genMapper(kryoWrapper: KryoSerializationWrapper[(Foo => Bar)])
               (foo: Foo) : Bar = {
    kryoWrapper.value.apply(foo)
}
val mapper = genMapper(KryoSerializationWrapper(new Blah(abc))) _
rdd.flatMap(mapper).collectAsMap()

object Blah(abc: ABC) extends (Foo => Bar) {
    def apply(foo: Foo) : Bar = { //This is the real function }
}

Siéntase libre de hacer Bla tan complicado como desee, clase, objeto compañero, clases anidadas, referencias a múltiples 3rd party libs.

KryoSerializationWrapper se refiere a: https://github.com/amplab/shark/blob/master/src/main/scala/shark/execution/serialization/KryoSerializationWrapper.scala

No estoy completamente seguro de que esto se aplique a Scala, pero, en Java, resolví el NotSerializableException refactorizando mi código para que el cierre no accediera a un campo final no serializable.

Me enfrenté a un problema similar, y lo que entiendo de La respuesta de Grega es

object NOTworking extends App {
 new testing().doIT
}
//adding extends Serializable wont help
class testing {

val list = List(1,2,3)

val rddList = Spark.ctx.parallelize(list)

def doIT =  {
  //again calling the fucntion someFunc 
  val after = rddList.map(someFunc(_))
  //this will crash (spark lazy)
  after.collect().map(println(_))
}

def someFunc(a:Int) = a+1

}

Su doIT método está tratando de serializar someFunc(_) método, sino como método no son serializables, intenta serializar clase prueba que no es serializable.

Así que haz que tu código funcione, debes definir someFunc dentro del método doIT. Por ejemplo:

def doIT =  {
 def someFunc(a:Int) = a+1
  //function definition
 }
 val after = rddList.map(someFunc(_))
 after.collect().map(println(_))
}

Y si hay múltiples funciones que entran en la imagen, entonces todos esas funciones deben estar disponibles para el contexto padre.