org.apache.spark.SparkException: Nhiệm vụ không serializable

Question

Đây là một ví dụ mã làm việc:

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

tôi nhận được báo lỗi dưới đây:

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

Vì bạn đang xác định chức năng bản đồ của bạn sử dụng một lớp bên trong vô danh, lớp có chứa cũng phải được Serializable. Xác định chức năng bản đồ của bạn như là một lớp riêng biệt hoặc làm cho nó trở thành một lớp bên trong tĩnh. Từ các tài liệu 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

chỉ cung cấp các mẫu mã:

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

khai báo lớp bên trong như là một biến tĩnh:

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