Spark 因 SerializedLambda 的 ClassNotFoundException 而失败

Question

我正在学习 spark，我的第一个程序是一个单词计数器，我在独立模式下执行该程序时遇到了问题

当我在本地模式下执行代码时，我没有任何问题 (setMaster("local")) 但是当我尝试在独立模式下（使用主从模式）执行时，它没有完成（我有运行脚本./sbin/start-all.sh)。该程序运行s。但是收集数据时，会发生异常。（在行 counter.collectAsMap() ）

错误如下：

15/01/30 15:23:54 WARN TaskSetManager: Lost task 1.0 in stage 0.0 (TID 1, ******): java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.spark.api.java.JavaRDDLike$$anonfun$fn.f of type org.apache.spark.api.java.function.FlatMapFunction in instance of org.apache.spark.api.java.JavaRDDLike$$anonfun$fn

代码如下：

public class MainTest {
    public static void main( String[] args ) throws {
        String origin = "originPath";

        SparkConf conf = new SparkConf().setAppName("org.sparkexample.WordCount").setMaster("spark://localhost:7077");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> file = context.textFile(origin);
        JavaRDD<String> words = file.flatMap(s -> Arrays.asList(s.split(" ")));
        JavaPairRDD<String, Integer> pairs = words.mapToPair(word ->new Tuple2<>(word,1));
        JavaPairRDD<String, Integer> counter = pairs.reduceByKey(( valueA,valueB) ->valueA +valueB);
        Map<String,Integer> map = counter.collectAsMap();
        map.forEach((key,value)->System.out.println(key+"="+value));
    }
}

我的机器安装了 Oracle Java 8，我用 ./sbin/start-all.sh 脚本启动主服务器和一个从服务器

哪一个可能是错误的？

更新：

我正在尝试 Java 7 实施，但我还有其他问题：例外情况是：

15/01/30 12:47:21 WARN TaskSetManager: Lost task 1.0 in stage 0.0 (TID 1, ***): java.lang.ClassNotFoundException: com.c4om.l3p3.sparkTest.MainTest

也许这是配置问题？

Java7代码如下：

public static void countWordsJava7(String path,Boolean local, Boolean printResult){
    SparkConf conf = new SparkConf().setAppName("org.sparkexample.WordCount").setMaster("spark://localhost:7077");
    JavaSparkContext context = new JavaSparkContext(conf);
    JavaRDD<String> file = context.textFile(path);
    JavaRDD<String> words = file.flatMap(new FlatMapFunction<String, String>() {
        @Override
        public Iterable<String> call(String s) {
            return Arrays.asList(" ".split(s));
        }
    });
    JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
        @Override
        public Tuple2<String, Integer> call(String s) {
            return new Tuple2<String, Integer>(s, 1);
        }
    });
    JavaPairRDD<String, Integer> counter = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
        @Override
        public Integer call(Integer i1, Integer i2) {
            return i1 + i2;
        }
    });
    Map<String, Integer> map = counter.collectAsMap();
}

Answer 1

ClassNotFoundException 表示工人缺少 class。可能有不同的原因，具体取决于 class 它抱怨的内容：

这是来自您自己的包的class。您还没有将包含您的代码的 jar 文件部署给工作人员。您可以使用 SparkConf.setJars 让 Spark 为您分发您的 jar。
这是一个系统 class。 这表明工作人员运行与 Java 的版本不同应用。例如 java.lang.invoke.SerializedLambda 是 Java 8 class — 如果缺少，则您的工人是运行 Java 7.
这是一个库class.你需要在所有worker上安装这个库，或者使用SparkConf.setJars让Spark分发你的罐子给你。

要调试此类问题，请在工作人员上打印 class路径。

rdd.foreachPartition { p => println(System.getProperty("java.class.path")) }

Answer 2

如果我运行从 IDE 像 Idea 一样编写代码，并调用 setJars(new String[]{"/path/to/jar/with/your/class.jar"，我会遇到同样的错误 cannot assign instance of java.lang.invoke.SerializedLambda }) 在 SparkConf 实例上，然后它起作用了。

例如：

SparkConf conf = new SparkConf().setAppName("LineCount");
conf.setMaster("spark://localhost:7077")
conf.setJars(new String[] { "/home/path/MainTest.jar" });

Answer 3

在 SparkConf 上设置 jar 路径然后它就可以工作了。 jar路径为maven或ant编译后的jar文件路径。

String master = "spark://localhost:7077";
    SparkConf conf = new SparkConf()
            .setAppName(WordCountTask.class.getName())
            .setMaster(master)
            .setJars(new String[]{"/home/user/Projects/spark-test/target/first-example-1.0-SNAPSHOT.jar"})

希望对您有所帮助。

Spark 因 SerializedLambda 的 ClassNotFoundException 而失败

Spark fails with ClassNotFoundException for SerializedLambda

java

apache-spark