java - Spark NotSerializableException

Question

在我的 Spark 代码中，我试图从一个 csv 文件创建一个 IndexedRowMatrix。但是，我收到以下错误:

Exception in thread "main" org.apache.spark.SparkException: Task not serializable
...
Caused by: java.io.NotSerializableException: org.apache.spark.api.java.JavaSparkContext

这是我的代码:

sc = new JavaSparkContext("local", "App",
              "/srv/spark", new String[]{"target/App.jar"});

JavaRDD<String> csv = sc.textFile("data/matrix.csv").cache();


JavaRDD<IndexedRow> entries = csv.zipWithIndex().map(
              new  Function<scala.Tuple2<String, Long>, IndexedRow>() {
                /**
                 * 
                **/ 
                private static final long serialVersionUID = 4795273163954440089L;

                @Override
                public IndexedRow call(Tuple2<String, Long> tuple)
                        throws Exception {
                    String line = tuple._1;
                    long index = tuple._2;
                    String[] strings = line.split(",");
                    double[] doubles = new double[strings.length];
                     for (int i = 0; i < strings.length; i++) {
                         doubles[i] = Double.parseDouble(strings[i]);
                     }
                     Vector v = new DenseVector(doubles);
                     return new IndexedRow(index, v);
                }
            });

Answer 1

我遇到了同样的问题。这让我陷入了困境。这是对匿名实例和可序列化性的 Java 限制。我的解决方案是将 Function 的匿名实例声明为实现 Serializable 的命名静态类并实例化它。我基本上声明了一个函数库，它是一个外部类，其中包含我想使用的函数的静态内部类定义。

当然，如果您用 Scala 编写它，它很可能是一个包含更简洁代码的文件，但在这种情况下这对您没有帮助。