org.apache.spark.SparkException: Task non serializzabile

Question

Questo è un esempio di codice di lavoro:

JavaPairDStream<String, String> messages = KafkaUtils.createStream(javaStreamingContext, zkQuorum, group, topicMap); 
messages.print(); 
JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() { 
    @Override 
    public String call(Tuple2<String, String> tuple2) { 
     return tuple2._2(); 
    } 
}); 

ottengo l'errore qui sotto:

ERROR: 
org.apache.spark.SparkException: Task not serializable 
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166) 
    at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158) 
    at org.apache.spark.SparkContext.clean(SparkContext.scala:1435) 
    at org.apache.spark.streaming.dstream.DStream.map(DStream.scala:438) 
    at org.apache.spark.streaming.api.java.JavaDStreamLike$class.map(JavaDStreamLike.scala:140) 
    at org.apache.spark.streaming.api.java.JavaPairDStream.map(JavaPairDStream.scala:46) 

Answer 1

Dal momento che si sta definendo la vostra funzione mappa utilizzando una classe interna anonima, la classe contenente deve anche essere serializzabile. Definisci la tua funzione mappa come una classe separata o rendila una classe interiore statica. Dalla documentazione di Java (http://docs.oracle.com/javase/8/docs/platform/serialization/spec/serial-arch.html):

Note - Serialization of inner classes (i.e., nested classes that are not static member classes), including local and anonymous classes, is strongly discouraged for several reasons. Because inner classes declared in non-static contexts contain implicit non-transient references to enclosing class instances, serializing such an inner class instance will result in serialization of its associated outer class instance as well.

Answer 2

solo fornendo il codice di esempio:

JavaDStream<String> lines = messages.map(mapFunc); 

dichiarare la classe interna come una variabile statica:

static Function<Tuple2<String, String>, String> mapFunc=new Function<Tuple2<String, String>, String>() { 
    @Override 
    public String call(Tuple2<String, String> tuple2) { 
     return tuple2._2(); 
    } 
}