java + spark：org.apache.spark.SparkException：作业中止：任务不可序列化：java.io.NotSerializableException答案

【问题标题】：java+spark: org.apache.spark.SparkException: Job aborted: Task not serializable: java.io.NotSerializableExceptionjava + spark：org.apache.spark.SparkException：作业中止：任务不可序列化：java.io.NotSerializableException
【发布时间】：2014-06-04 20:14:31
【问题描述】：

我是 spark 新手，正在尝试运行示例 JavaSparkPi.java，它运行良好，但是因为我必须在另一个 java 中使用它，所以我将所有内容从 main 复制到类中的方法并尝试在main中调用方法，它说

org.apache.spark.SparkException：作业中止：任务不可序列化： java.io.NotSerializableException

代码如下所示：

public class JavaSparkPi {

public void cal(){
    JavaSparkContext jsc = new JavaSparkContext("local", "JavaLogQuery");
    int slices = 2;
    int n = 100000 * slices;

    List<Integer> l = new ArrayList<Integer>(n);
    for (int i = 0; i < n; i++) {
        l.add(i);
    }

    JavaRDD<Integer> dataSet = jsc.parallelize(l, slices);

    System.out.println("count is: "+ dataSet.count());
    dataSet.foreach(new VoidFunction<Integer>(){
        public void call(Integer i){
            System.out.println(i);
        }
    });

    int count = dataSet.map(new Function<Integer, Integer>() {
        @Override
        public Integer call(Integer integer) throws Exception {
            double x = Math.random() * 2 - 1;
            double y = Math.random() * 2 - 1;
            return (x * x + y * y < 1) ? 1 : 0;
        }
    }).reduce(new Function2<Integer, Integer, Integer>() {
        @Override
        public Integer call(Integer integer, Integer integer2) throws Exception {
            return integer + integer2;
        }
    });

    System.out.println("Pi is roughly " + 4.0 * count / n);
}

public static void main(String[] args) throws Exception {

    JavaSparkPi myClass = new JavaSparkPi();
    myClass.cal();
}
}

有人对此有想法吗？谢谢！

【问题讨论】：

您省略了错误中最重要的一行。什么类不可序列化？
@SeanOwen 线程“主”org.apache.spark.SparkException 中的异常：作业中止：任务不可序列化：java.io.NotSerializableException：com.mycompany.sparkproject2.JavaSparkPi 是这个吗？
我认为您的问题已经在这里得到解答：stackoverflow.com/questions/22592811/…
Task not serializable: java.io.NotSerializableException when calling function outside closure only on classes not objects的可能重复

标签： java serialization apache-spark

【解决方案1】：

嵌套函数持有对包含对象 (JavaSparkPi) 的引用。所以这个对象将被序列化。为此，它需要可序列化。做起来很简单：

public class JavaSparkPi implements Serializable {
  ...

【讨论】：

这是对的，或者，如果您不能或不想将外部类设为静态，则将匿名类设为内部静态类。不过，这在代码方面更麻烦
stackoverflow.com/questions/40818001/… 理解 spark 序列化的概述

【解决方案2】：

主要问题是，当您在 java 中创建匿名类时，它会传递封闭类的引用。这可以通过多种方式解决

声明封闭类Serializable

这适用于你的情况，但如果你的封闭类有一些不可序列化的字段，它就会失败。我还要说，序列化父类完全是浪费。

在静态函数中创建闭包

通过调用一些静态函数来创建闭包不会将引用传递给闭包，因此不需要以这种方式进行序列化。

【讨论】：

【解决方案3】：

出现此错误是因为您的本地或集群中有多个物理 CPU，并且 spark 引擎尝试通过网络将此函数发送到多个 CPU。你的功能

 dataSet.foreach(new VoidFunction<Integer>(){
        public void call(Integer i){
            ***System.out.println(i);***
        }
    });

使用未序列化的 println()。所以 Spark Engine 抛出的异常。解决方案是您可以在下面使用：

dataSet.collect().forEach(new VoidFunction<Integer>(){
       public void call(Integer i){
         System.out.println(i);
    }
});

【讨论】：