La mappa non può essere serializzabile in scala?

Question

Sono nuovo di Scala. Come mai la funzione "mappa" non è serializzabile? Come renderlo serializzabile? Ad esempio, se il mio codice è come qui di seguito:

val data = sc.parallelize(List(1,4,3,5,2,3,5)) 

def myfunc(iter: Iterator[Int]) : Iterator[Int] = { 
    val lst = List(("a", 1),("b", 2),("c",3), ("a",2)) 
    var res = List[Int]() 
    while (iter.hasNext) { 
    val cur = iter.next 
    val a = lst.groupBy(x => x._1).mapValues(_.size) 
    //val b= a.map(x => x._2) 
    res = res ::: List(cur) 
    } 
    res.iterator 
} 

data.mapPartitions(myfunc).collect 

Se io commento dalla riga

val b= a.map(x => x._2) 

Il codice restituisce un'eccezione:

org.apache.spark.SparkException: Task not serializable 
Caused by: java.io.NotSerializableException: scala.collection.immutable.MapLike$$anon$2 
Serialization stack: 
    - object not serializable (class: scala.collection.immutable.MapLike$$anon$2, value: Map(1 -> 3)) 
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: a, type: interface scala.collection.immutable.Map) 

Grazie mille.

Answer 1

E 'ben noto bug Scala: https://issues.scala-lang.org/browse/SI-7005 Mappa # mapValues ​​non è serializzabile

abbiamo questo problema nelle nostre applicazioni Spark, map(identity) risolve il problema

rdd.groupBy(_.segment).mapValues(v => ...).map(identity) 

Answer 2

Hai provato l'esecuzione di questo stesso codice in un applicazione? Sospetto che questo sia un problema con la scintilla. Se si vuole farlo funzionare nel guscio scintilla allora si potrebbe provare avvolgendo la definizione di myfunc e la sua applicazione tra parentesi graffe in questo modo:

val data = sc.parallelize(List(1,4,3,5,2,3,5)) 

val result = { 
    def myfunc(iter: Iterator[Int]) : Iterator[Int] = { 
    val lst = List(("a", 1),("b", 2),("c",3), ("a",2)) 
    var res = List[Int]() 
    while (iter.hasNext) { 
     val cur = iter.next 
     val a = lst.groupBy(x => x._1).mapValues(_.size) 
     val b= a.map(x => x._2) 
     res = res ::: List(cur) 
    } 
    res.iterator 
    } 
    data.mapPartitions(myfunc).collect 
} 

Answer 3

L'effettiva attuazione della funzione mapValues ​​è fornito qui sotto e come si può vedere non è serializzabile e crea solo una vista non una corretta esistenza dei dati e quindi si sta ottenendo questo errore. I mapValues ​​della situazione hanno molti vantaggi.

protected class MappedValues[C](f: B => C) extends AbstractMap[A, C] with DefaultMap[A, C] { 
override def foreach[D](g: ((A, C)) => D): Unit = for ((k, v) <- self) g((k, f(v))) 
def iterator = for ((k, v) <- self.iterator) yield (k, f(v)) 
override def size = self.size 
override def contains(key: A) = self.contains(key) 
def get(key: A) = self.get(key).map(f) 
}