Zadanie nie serializowalne: java. io. NotSerializableException podczas wywoływania funkcji poza zamknięciem tylko na klasach nie obiektach

Odpowiedź Grega jest świetna w wyjaśnianiu, dlaczego oryginalny kod nie działa i na dwa sposoby rozwiązania problemu. Jednak to rozwiązanie nie jest zbyt elastyczne; rozważ przypadek, w którym Twoje zamknięcie zawiera wywołanie metody na klasie nie - Serializable, nad którą nie masz kontroli. Nie można dodać znacznika Serializable do tej klasy ani zmienić implementacji bazowej, aby zmienić metodę w funkcję.

Nilesh przedstawia świetne obejście tego problemu, ale rozwiązaniem może być bardziej zwięzłe i ogólne:

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

Ta funkcja-serializer może być następnie używana do automatycznego zawijania zamknięć i wywołań metod:

rdd map genMapper(someFunc)

Complete talk w pełni wyjaśniające problem, który proponuje świetny sposób zmiany paradygmatu, aby uniknąć problemów z serializacją: https://github.com/samthebest/dump/blob/master/sams-scala-tutorial/serialization-exceptions-and-memory-leaks-no-ws.md

Najczęściej głosowana odpowiedź w zasadzie sugeruje wyrzucenie całej funkcji języka - czyli nie używania już metod, a tylko funkcji. Rzeczywiście w programowaniu funkcyjnym metody w klasach należy unikać, ale przekształcenie ich w funkcje nie rozwiązuje tutaj problemu projektowego(patrz powyższy link).

Jako szybkie rozwiązanie w tej konkretnej sytuacji możesz po prostu użyć adnotacji @transient, aby powiedzieć jej, aby nie próbowała serializować obrażającej wartości (tutaj Spark.ctx jest niestandardową klasą, A nie klasą Spark po nazwie OP):

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

Możesz również zrestrukturyzować kod tak, aby rddlist mieszkał gdzie indziej, ale to też jest paskudne.

Przyszłość to prawdopodobnie zarodniki

W przyszłości Scala będzie zawierać te rzeczy zwane "zarodnikami", które powinny pozwolić nam drobnoziarniste kontrolowanie tego, co robi, a nie dokładnie zostaje wciągnięte przez zamknięcie. Co więcej, powinno to zmienić wszystkie błędy przypadkowego ściągania typów nieserializowalnych (lub niechcianych wartości) w błędy kompilacji, a nie teraz, co jest strasznym wyjątkami / wyciekami pamięci.

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

Wskazówka na temat serializacji Kryo

Używając kyro, zrób to tak, aby Rejestracja jest konieczna, oznacza to, że zamiast wycieków pamięci pojawią się Błędy:

Strategia rejestracji zajęć z kryo

Oczywiście daje to tylko kontrolę na poziomie typu, a nie na poziomie wartości Kontrola.

... więcej pomysłów na przyszłość.

Rozwiązałem ten problem używając innego podejścia. Po prostu musisz serializować obiekty przed przejściem przez zamknięcie, a następnie usunąć serializację. To podejście po prostu działa, nawet jeśli twoje zajęcia nie są Serializowalne, ponieważ wykorzystuje Kryo za kulisami. Wszystko czego potrzebujesz to curry. ;)

Oto przykład jak to zrobiłem:

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

Możesz zrobić Bla tak skomplikowane, jak chcesz, Klasa, obiekt towarzyszący, zagnieżdżone klasy, odniesienia do wielu stron trzecich libs.

KryoSerializationWrapper odnosi się do: https://github.com/amplab/shark/blob/master/src/main/scala/shark/execution/serialization/KryoSerializationWrapper.scala

Nie jestem do końca pewien, czy dotyczy to Scali, ale w Javie rozwiązałem NotSerializableException poprzez refaktoryzację kodu tak, aby zamknięcie nie uzyskało dostępu do niezwiązanego z serializowalnym polem final.

Miałem do czynienia z podobnym problemem, a to, co rozumiem z odpowiedzi Grega to

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

}

Twoja metodadoIT próbuje serializowaćmetodę someFunc (_) , ale ponieważ metody nie są serializowalne, próbuje serializować klasętesting , która ponownie nie jest serializowalna.

Aby Twój kod działał, powinieneś zdefiniować someFuncwewnątrz metody doIT. Na przykład:

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

A Jeśli pojawi się wiele funkcji, to wszystkie funkcje te powinny być dostępne dla kontekstu nadrzędnego.