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

【问题标题】：org.apache.spark.SparkException: Task not serializable - JavaSparkContextorg.apache.spark.SparkException：任务不可序列化 - JavaSparkContext
【发布时间】：2015-08-22 22:53:46
【问题描述】：

我正在尝试运行以下简单的 Spark 代码：

Gson gson = new Gson();
JavaRDD<String> stringRdd = jsc.textFile("src/main/resources/META-INF/data/supplier.json");

JavaRDD<SupplierDTO> rdd = stringRdd.map(new Function<String, SupplierDTO>()
{
    private static final long serialVersionUID = -78238876849074973L;

    @Override
    public SupplierDTO call(String str) throws Exception
    {
        return gson.fromJson(str, SupplierDTO.class);
    }
});

但在执行stringRdd.map 语句时抛出以下错误：

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:1478)
at org.apache.spark.rdd.RDD.map(RDD.scala:288)
at org.apache.spark.api.java.JavaRDDLike$class.map(JavaRDDLike.scala:78)
at org.apache.spark.api.java.JavaRDD.map(JavaRDD.scala:32)
at com.demo.spark.processor.cassandra.CassandraDataUploader.uploadData(CassandraDataUploader.java:71)
at com.demo.spark.processor.cassandra.CassandraDataUploader.main(CassandraDataUploader.java:47)
Caused by: java.io.NotSerializableException: org.apache.spark.api.java.JavaSparkContext
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:42)
at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:73)
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164)
... 7 more

这里的 'jsc' 是我正在使用的 JavaSparkContext 对象。据我所知，JavaSparkContext 不是Serializable 对象，不应在将发送给 Spark 工作人员的任何函数中使用它。

现在，我无法理解的是，JavaSparkContext 的实例是如何发送给工作人员的？为了避免这种情况，我应该在我的代码中进行哪些更改？

【问题讨论】：

你确定这就是所有的日志吗？我认为是 Gson 导致了问题，因为我之前遇到过同样的问题。
这是完整的日志。
你能把 SupplierDTO 类粘贴到这里吗？
@Sathish：这只是一个 pojo 和可序列化的..

标签： java serialization apache-spark

【解决方案1】：

gson 引用将外部类“拉”到闭包的范围内，并带走其完整的对象图。

在这种情况下，在闭包中创建 gson 对象：

public SupplierDTO call(String str) throws Exception {   
   Gson gson = Gson();
   return gson.fromJson(str, SupplierDTO.class);
}

你也可以声明火花上下文transient

如果创建 Gson 实例的成本很高，请考虑使用mapPartitions 而不是map。

【讨论】：

在 call 方法中创建 Gson 对象没有任何区别。您提到的其他两个选项都不是。请帮忙。
@ArkaGhosh 将 transient 添加到 sparkContext 的声明应该 - 至少 - 改变被抛出的异常。
能否请您提及该transient 注释的完全分类名称。
@ArkaGhosh transient 是 Java 中的关键字。你像这样使用它：transient JavaSparkContext context = new ....
我的错.. 我以为你在谈论 transient 注释.. 让我再试一次..

【解决方案2】：

对我来说，我使用以下选项之一解决了这个问题：

如上所述，通过将 SparkContext 声明为 transient
您也可以尝试将对象 gson 设为静态 static Gson gson = new Gson();

请参考文档Job aborted due to stage failure: Task not serializable

查看解决此问题的其他可用选择

【讨论】：

我也有同样的问题。我将我的上下文参考标记为瞬态。我工作过

【解决方案3】：

您可以使用下面的代码代替第 9 行。(return gson.fromJson(str, SupplierDTO.class);)

return new Gson().fromJson(str, SupplierDTO.class);//this is correct

并删除第 1 行。(Gson gson = new Gson();)

【讨论】：