Il processo di spark non è riuscito a causa di java.io.NotSerializableException: org.apache.spark.SparkContext

Question

Sto affrontando un'eccezione sopra quando sto tentando di applicare un metodo (ComputeDwt) sull'input RDD[(Int,ArrayBuffer[(Int,Double)])]. Sto anche usando l'opzione extends Serialization per serializzare gli oggetti nella scintilla. Ecco lo snippet di codice.

input:series:RDD[(Int,ArrayBuffer[(Int,Double)])] 
DWTsample extends Serialization is a class having computeDwt function. 
sc: sparkContext 

val kk:RDD[(Int,List[Double])]=series.map(t=>(t._1,new DWTsample().computeDwt(sc,t._2))) 

Error: 
org.apache.spark.SparkException: Job failed: java.io.NotSerializableException: org.apache.spark.SparkContext 
org.apache.spark.SparkException: Job failed: java.io.NotSerializableException: org.apache.spark.SparkContext 
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:760) 
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:758) 
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:60) 
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) 
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:758) 
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$submitMissingTasks(DAGScheduler.scala:556) 
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$submitStage(DAGScheduler.scala:503) 
at org.apache.spark.scheduler.DAGScheduler.processEvent(DAGScheduler.scala:361) 
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$run(DAGScheduler.scala:441) 
at org.apache.spark.scheduler.DAGScheduler$$anon$1.run(DAGScheduler.scala:149) 

Qualcuno potrebbe suggerirmi quale potrebbe essere il problema e cosa si dovrebbe fare per risolvere questo problema?

Answer 1

La linea

series.map(t=>(t._1,new DWTsample().computeDwt(sc,t._2))) 

riferimenti SparkContext (sc) ma SparkContext non è serializzabile. SparkContext è progettato per esporre le operazioni eseguite sul driver; non può essere referenziato/usato dal codice che viene eseguito sui lavoratori.

È necessario ristrutturare il codice in modo che sc non faccia riferimento alla chiusura della funzione mappa.