org.apache.spark.SparkException：任务不可序列化答案

【问题标题】：org.apache.spark.SparkException: Task not serializableorg.apache.spark.SparkException：任务不可序列化
【发布时间】：2015-05-31 11:53:02
【问题描述】：

这是一个工作代码示例：

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

我收到以下错误：

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)

【问题讨论】：

好吧，如果它正在工作，那就太好了:)。如果不是，可以用-Dsun.io.serialization.extendedDebugInfo=true开启Java序列化调试。
谢谢，不行，我试过了。 JavaDStream lines = messages.map(new Function, String>() { @Override public String call(Tuple2 tuple2) { return tuple2._2(); } }) ;这行代码有问题。
很确定这段代码是 Java 而不是 Scala（即标签）

标签： scala apache-spark apache-kafka

【解决方案1】：

由于您使用匿名内部类定义地图函数，因此包含的类也必须是可序列化的。将您的地图函数定义为单独的类或使其成为静态内部类。来自 Java 文档 (http://docs.oracle.com/javase/8/docs/platform/serialization/spec/serial-arch.html)：

注意 - 强烈建议不要序列化内部类（即不是静态成员类的嵌套类），包括本地和匿名类，原因有几个。因为在非静态上下文中声明的内部类包含对封闭类实例的隐式非瞬态引用，所以序列化这样的内部类实例也会导致其关联的外部类实例的序列化。

【讨论】：

很高兴它有帮助！如果有请采纳答案
Spark 正在尝试序列化传递给 map 的对象，但由于未实现 Serializable 而无法对其进行序列化？为什么 Spark 要做序列化？如果我们将 map 函数定义为一个单独的类，我们是否也需要将其设为 Serializable ？

【解决方案2】：

只提供代码示例：

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

将内部类声明为静态变量：

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

【讨论】：