Org.apache.spark.SparkException: task is not serializable

Question

This is an example of working code:

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(); } }); 

I get the following error:

 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

Since you define your map function using an anonymous inner class, the containing class must also be Serializable. Define your map function as a separate class or make it a static inner class. From the Java documentation ( http://docs.oracle.com/javase/8/docs/platform/serialization/spec/serial-arch.html ):

Note. Serializing 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-cross-reference references to the inclusion of class instances, serializing such an inner class instance will result in the serialization of the associated outer class instance.

Answer 2

just enter the sample code:

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

declare the inner class as a static variable:

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