Spark wordcount 断言失败：不安全符号不稳定

Question

我已经使用 Hortonworks Ambari 安装了 HDFS、Yarn 和 Spark。我已经编写了简单的程序 read/read 到 HDFS，Map-Reduce wordcount，一切正常。

然后我尝试测试 Spark。我从官方Spark示例中复制了字数统计程序：

public final class JavaWordCount {
  private static final Pattern SPACE = Pattern.compile(" ");

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

    if (args.length < 1) {
      System.err.println("Usage: JavaWordCount <file>");
      System.exit(1);
    }

    SparkSession spark = SparkSession
      .builder()
      .appName("JavaWordCount")
      .getOrCreate();

    JavaRDD<String> lines = spark.read().textFile(args[0]).javaRDD();

    JavaRDD<String> words = lines.flatMap(s -> Arrays.asList(SPACE.split(s)).iterator());

    JavaPairRDD<String, Integer> ones = words.mapToPair(s -> new Tuple2<>(s, 1));

    JavaPairRDD<String, Integer> counts = ones.reduceByKey((i1, i2) -> i1 + i2);

    List<Tuple2<String, Integer>> output = counts.collect();
    for (Tuple2<?,?> tuple : output) {
      System.out.println(tuple._1() + ": " + tuple._2());
    }
    spark.stop();
  }
}

我在 hdfs://server:8020/user/root/input.txt.user/root/input.txt.input.txt 中放置了一个内容为 "Hello World" 的文件。

然后我运行这个程序使用下面的命令，

java -cp  "all necessary jars provided by Ambari, I believe the version should match" JavaWordCount hdfs://server:8020/user/root/input.txt

然后有一个非常奇怪的异常，即使我搜索 google 我也找不到任何相似的匹配项。

异常发生在第 27 行，即

JavaRDD<String> lines = spark.read().textFile(args[0]).javaRDD();

而且很奇怪，我在搜索引擎上找不到任何完全匹配的异常。

是否有任何提示可供我检查和研究以使其正常工作？谢谢。例外很长，最后列出来

java.lang.AssertionError: assertion failed: unsafe symbol Unstable (child of package InterfaceStability) in runtime reflection universe
        at scala.reflect.internal.Symbols$Symbol.<init>(Symbols.scala:184)
        at scala.reflect.internal.Symbols$TypeSymbol.<init>(Symbols.scala:3009)
        at scala.reflect.internal.Symbols$ClassSymbol.<init>(Symbols.scala:3201)
        at scala.reflect.internal.Symbols$StubClassSymbol.<init>(Symbols.scala:3496)
        at scala.reflect.internal.Symbols$Symbol.newStubSymbol(Symbols.scala:498)
        at scala.reflect.internal.pickling.UnPickler$Scan.readExtSymbol(UnPickler.scala:258)
        at scala.reflect.internal.pickling.UnPickler$Scan.readSymbol(UnPickler.scala:284)
        at scala.reflect.internal.pickling.UnPickler$Scan.readSymbolRef(UnPickler.scala:649)
        at scala.reflect.internal.pickling.UnPickler$Scan.readType(UnPickler.scala:417)
        at scala.reflect.internal.pickling.UnPickler$Scan$$anonfun$readTypeRef.apply(UnPickler.scala:658)
        at scala.reflect.internal.pickling.UnPickler$Scan$$anonfun$readTypeRef.apply(UnPickler.scala:658)
        at scala.reflect.internal.pickling.UnPickler$Scan.at(UnPickler.scala:179)
        at scala.reflect.internal.pickling.UnPickler$Scan.readTypeRef(UnPickler.scala:658)
        at scala.reflect.internal.pickling.UnPickler$Scan.readAnnotationInfo(UnPickler.scala:492)
        at scala.reflect.internal.pickling.UnPickler$Scan.readSymbolAnnotation(UnPickler.scala:515)
        at scala.reflect.internal.pickling.UnPickler$Scan.run(UnPickler.scala:97)
        at scala.reflect.internal.pickling.UnPickler.unpickle(UnPickler.scala:38)
        at scala.reflect.runtime.JavaMirrors$JavaMirror.unpickleClass(JavaMirrors.scala:619)
        at scala.reflect.runtime.SymbolLoaders$TopClassCompleter$$anonfun$complete.apply$mcV$sp(SymbolLoaders.scala:28)
        at scala.reflect.runtime.SymbolLoaders$TopClassCompleter$$anonfun$complete.apply(SymbolLoaders.scala:25)
        at scala.reflect.runtime.SymbolLoaders$TopClassCompleter$$anonfun$complete.apply(SymbolLoaders.scala:25)
        at scala.reflect.internal.SymbolTable.slowButSafeEnteringPhaseNotLaterThan(SymbolTable.scala:263)
        at scala.reflect.runtime.SymbolLoaders$TopClassCompleter.complete(SymbolLoaders.scala:25)
        at scala.reflect.internal.Symbols$Symbol.info(Symbols.scala:1514)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon.scala$reflect$runtime$SynchronizedSymbols$SynchronizedSymbol$$super$info(SynchronizedSymbols.scala:189)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info.apply(SynchronizedSymbols.scala:127)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info.apply(SynchronizedSymbols.scala:127)
        at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
        at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:123)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:189)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.info(SynchronizedSymbols.scala:127)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon.info(SynchronizedSymbols.scala:189)
        at scala.reflect.internal.SymbolTable.openPackageModule(SymbolTable.scala:286)
        at scala.reflect.internal.SymbolTable.openPackageModule(SymbolTable.scala:341)
        at scala.reflect.runtime.SymbolLoaders$LazyPackageType$$anonfun$complete.apply$mcV$sp(SymbolLoaders.scala:74)
        at scala.reflect.runtime.SymbolLoaders$LazyPackageType$$anonfun$complete.apply(SymbolLoaders.scala:71)
        at scala.reflect.runtime.SymbolLoaders$LazyPackageType$$anonfun$complete.apply(SymbolLoaders.scala:71)
        at scala.reflect.internal.SymbolTable.slowButSafeEnteringPhaseNotLaterThan(SymbolTable.scala:263)
        at scala.reflect.runtime.SymbolLoaders$LazyPackageType.complete(SymbolLoaders.scala:71)
        at scala.reflect.internal.Symbols$Symbol.info(Symbols.scala:1514)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon.scala$reflect$runtime$SynchronizedSymbols$SynchronizedSymbol$$super$info(SynchronizedSymbols.scala:174)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info.apply(SynchronizedSymbols.scala:127)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info.apply(SynchronizedSymbols.scala:127)
        at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
        at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:123)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:174)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.info(SynchronizedSymbols.scala:127)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon.info(SynchronizedSymbols.scala:174)
        at scala.reflect.internal.Types$TypeRef.thisInfo(Types.scala:2194)
        at scala.reflect.internal.Types$TypeRef.baseClasses(Types.scala:2199)
        at scala.reflect.internal.tpe.FindMembers$FindMemberBase.<init>(FindMembers.scala:17)
        at scala.reflect.internal.tpe.FindMembers$FindMember.<init>(FindMembers.scala:219)
        at scala.reflect.internal.Types$Type.scala$reflect$internal$Types$Type$$findMemberInternal(Types.scala:1014)
        at scala.reflect.internal.Types$Type.findMember(Types.scala:1016)
        at scala.reflect.internal.Types$Type.memberBasedOnName(Types.scala:631)
        at scala.reflect.internal.Types$Type.member(Types.scala:600)
        at scala.reflect.internal.Mirrors$RootsBase.getModuleOrClass(Mirrors.scala:48)
        at scala.reflect.internal.Mirrors$RootsBase.getModuleOrClass(Mirrors.scala:45)
        at scala.reflect.internal.Mirrors$RootsBase.getModuleOrClass(Mirrors.scala:66)
        at scala.reflect.internal.Mirrors$RootsBase.staticModuleOrClass(Mirrors.scala:77)
        at scala.reflect.internal.Mirrors$RootsBase.staticModule(Mirrors.scala:161)
        at scala.reflect.internal.Mirrors$RootsBase.staticModule(Mirrors.scala:22)
        at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$optionOfProductType$$typecreator46.apply(ScalaReflection.scala:654)
        at scala.reflect.api.TypeTags$WeakTypeTagImpl.tpe$lzycompute(TypeTags.scala:232)
        at scala.reflect.api.TypeTags$WeakTypeTagImpl.tpe(TypeTags.scala:232)
        at org.apache.spark.sql.catalyst.ScalaReflection$class.localTypeOf(ScalaReflection.scala:839)
        at org.apache.spark.sql.catalyst.ScalaReflection$.localTypeOf(ScalaReflection.scala:39)
        at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$optionOfProductType.apply$mcZ$sp(ScalaReflection.scala:654)
        at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$optionOfProductType.apply(ScalaReflection.scala:653)
        at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$optionOfProductType.apply(ScalaReflection.scala:653)
        at scala.reflect.internal.tpe.TypeConstraints$UndoLog.undo(TypeConstraints.scala:56)
        at org.apache.spark.sql.catalyst.ScalaReflection$class.cleanUpReflectionObjects(ScalaReflection.scala:824)
        at org.apache.spark.sql.catalyst.ScalaReflection$.cleanUpReflectionObjects(ScalaReflection.scala:39)
        at org.apache.spark.sql.catalyst.ScalaReflection$.optionOfProductType(ScalaReflection.scala:652)
        at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$.apply(ExpressionEncoder.scala:51)
        at org.apache.spark.sql.Encoders$.STRING(Encoders.scala:96)
        at org.apache.spark.sql.SQLImplicits.newStringEncoder(SQLImplicits.scala:73)
        at org.apache.spark.sql.DataFrameReader.textFile(DataFrameReader.scala:732)
        at org.apache.spark.sql.DataFrameReader.textFile(DataFrameReader.scala:702)
        at JavaWordCount .main(JavaWordCount .java:27)
Exception in thread "main" java.lang.RuntimeException: error reading Scala signature of org.apache.spark.sql.package: assertion failed: unsafe symbol Unstable (child of package InterfaceStability) in runtime reflection universe
        at scala.reflect.internal.pickling.UnPickler.unpickle(UnPickler.scala:46)
        at scala.reflect.runtime.JavaMirrors$JavaMirror.unpickleClass(JavaMirrors.scala:619)
        at scala.reflect.runtime.SymbolLoaders$TopClassCompleter$$anonfun$complete.apply$mcV$sp(SymbolLoaders.scala:28)
        at scala.reflect.runtime.SymbolLoaders$TopClassCompleter$$anonfun$complete.apply(SymbolLoaders.scala:25)
        at scala.reflect.runtime.SymbolLoaders$TopClassCompleter$$anonfun$complete.apply(SymbolLoaders.scala:25)
        at scala.reflect.internal.SymbolTable.slowButSafeEnteringPhaseNotLaterThan(SymbolTable.scala:263)
        at scala.reflect.runtime.SymbolLoaders$TopClassCompleter.complete(SymbolLoaders.scala:25)
        at scala.reflect.internal.Symbols$Symbol.info(Symbols.scala:1514)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon.scala$reflect$runtime$SynchronizedSymbols$SynchronizedSymbol$$super$info(SynchronizedSymbols.scala:189)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info.apply(SynchronizedSymbols.scala:127)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info.apply(SynchronizedSymbols.scala:127)
        at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
        at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:123)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:189)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.info(SynchronizedSymbols.scala:127)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon.info(SynchronizedSymbols.scala:189)
        at scala.reflect.internal.SymbolTable.openPackageModule(SymbolTable.scala:286)
        at scala.reflect.internal.SymbolTable.openPackageModule(SymbolTable.scala:341)
        at scala.reflect.runtime.SymbolLoaders$LazyPackageType$$anonfun$complete.apply$mcV$sp(SymbolLoaders.scala:74)
        at scala.reflect.runtime.SymbolLoaders$LazyPackageType$$anonfun$complete.apply(SymbolLoaders.scala:71)
        at scala.reflect.runtime.SymbolLoaders$LazyPackageType$$anonfun$complete.apply(SymbolLoaders.scala:71)
        at scala.reflect.internal.SymbolTable.slowButSafeEnteringPhaseNotLaterThan(SymbolTable.scala:263)
        at scala.reflect.runtime.SymbolLoaders$LazyPackageType.complete(SymbolLoaders.scala:71)
        at scala.reflect.internal.Symbols$Symbol.info(Symbols.scala:1514)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon.scala$reflect$runtime$SynchronizedSymbols$SynchronizedSymbol$$super$info(SynchronizedSymbols.scala:174)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info.apply(SynchronizedSymbols.scala:127)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info.apply(SynchronizedSymbols.scala:127)
        at scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
        at scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:123)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:174)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.info(SynchronizedSymbols.scala:127)
        at scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon.info(SynchronizedSymbols.scala:174)
        at scala.reflect.internal.Types$TypeRef.thisInfo(Types.scala:2194)
        at scala.reflect.internal.Types$TypeRef.baseClasses(Types.scala:2199)
        at scala.reflect.internal.tpe.FindMembers$FindMemberBase.<init>(FindMembers.scala:17)
        at scala.reflect.internal.tpe.FindMembers$FindMember.<init>(FindMembers.scala:219)
        at scala.reflect.internal.Types$Type.scala$reflect$internal$Types$Type$$findMemberInternal(Types.scala:1014)
        at scala.reflect.internal.Types$Type.findMember(Types.scala:1016)
        at scala.reflect.internal.Types$Type.memberBasedOnName(Types.scala:631)
        at scala.reflect.internal.Types$Type.member(Types.scala:600)
        at scala.reflect.internal.Mirrors$RootsBase.getModuleOrClass(Mirrors.scala:48)
        at scala.reflect.internal.Mirrors$RootsBase.getModuleOrClass(Mirrors.scala:45)
        at scala.reflect.internal.Mirrors$RootsBase.getModuleOrClass(Mirrors.scala:66)
        at scala.reflect.internal.Mirrors$RootsBase.staticModuleOrClass(Mirrors.scala:77)
        at scala.reflect.internal.Mirrors$RootsBase.staticModule(Mirrors.scala:161)
        at scala.reflect.internal.Mirrors$RootsBase.staticModule(Mirrors.scala:22)
        at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$optionOfProductType$$typecreator46.apply(ScalaReflection.scala:654)
        at scala.reflect.api.TypeTags$WeakTypeTagImpl.tpe$lzycompute(TypeTags.scala:232)
        at scala.reflect.api.TypeTags$WeakTypeTagImpl.tpe(TypeTags.scala:232)
        at org.apache.spark.sql.catalyst.ScalaReflection$class.localTypeOf(ScalaReflection.scala:839)
        at org.apache.spark.sql.catalyst.ScalaReflection$.localTypeOf(ScalaReflection.scala:39)
        at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$optionOfProductType.apply$mcZ$sp(ScalaReflection.scala:654)
        at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$optionOfProductType.apply(ScalaReflection.scala:653)
        at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$optionOfProductType.apply(ScalaReflection.scala:653)
        at scala.reflect.internal.tpe.TypeConstraints$UndoLog.undo(TypeConstraints.scala:56)
        at org.apache.spark.sql.catalyst.ScalaReflection$class.cleanUpReflectionObjects(ScalaReflection.scala:824)
        at org.apache.spark.sql.catalyst.ScalaReflection$.cleanUpReflectionObjects(ScalaReflection.scala:39)
        at org.apache.spark.sql.catalyst.ScalaReflection$.optionOfProductType(ScalaReflection.scala:652)
        at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$.apply(ExpressionEncoder.scala:51)
        at org.apache.spark.sql.Encoders$.STRING(Encoders.scala:96)
        at org.apache.spark.sql.SQLImplicits.newStringEncoder(SQLImplicits.scala:73)
        at org.apache.spark.sql.DataFrameReader.textFile(DataFrameReader.scala:732)
        at org.apache.spark.sql.DataFrameReader.textFile(DataFrameReader.scala:702)
        at JavaWordCount .main(JavaWordCount .java:27)

Answer 1

尝试使用spark-submit脚本提交spark作业，不能直接使用java命令，虽然spark-submit也使用java命令。

Spark wordcount 断言失败：不安全符号不稳定

Spark wordcount assertion failed: unsafe symbol Unstable

java

apache-spark

hadoop

ambari