java - Spark 因 SerializedLambda 的 ClassNotFoundException 失败

Question

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

当我在本地模式下执行代码时，没有任何问题(setMaster("local"))，但是当我尝试独立执行(使用主从)时，它没有完成(我已经运行了脚本./sbin/start-all.sh)。程序运行。但收集数据时，出现异常。 (在 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$1$1.f$3 of type org.apache.spark.api.java.function.FlatMapFunction in instance of org.apache.spark.api.java.JavaRDDLike$$anonfun$fn$1$1

代码如下:

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 实现，但有其他问题:异常(exception)是:

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 表示工作人员缺少类。可能有不同的原因，具体取决于它所提示的类:

• 这是您自己的包中的类。您尚未将包含代码的 jar 文件部署到工作线程。您可以使用 SparkConf.setJars 让 Spark 为您分发 jar。

• 它是一个系统类。这表明工作线程运行的 Java 版本与应用程序不同。例如，java.lang.invoke.SerializedLambda 是一个 Java 8 类 - 如果缺少该类，则您的工作线程正在运行 Java 7。

• 它是一个库类。您需要在所有工作线程上安装该库，或者使用 SparkConf.setJars 让 Spark 为您分发您的 jar。

要调试此类问题，请打印工作线程上的类路径。

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