使用spark/livy的todf时发生nosuchelementexception

qlzsbp2j  于 2021-06-09  发布在  Hbase
关注(0)|答案(1)|浏览(475)

我试图从spark中生成一个sparkDataframe，它已经使用apachelivy进行了初始化。
我第一次在这个更复杂的hbase调用中注意到这个问题：

import spark.implicits._

 ... 

        spark.sparkContext
          .newAPIHadoopRDD(
            conf,
            classOf[TableInputFormat],
            classOf[ImmutableBytesWritable],
            classOf[Result]
          )
          .toDF()

但我发现我可以在一个简单的

import spark.implicits._

  ...

  val filtersDf = filters.toDF()

哪里， filtersDf 只是一系列案例类。
共同的问题是 *.toDF() 但是，它也发生在 *.toDS() ，这让我觉得 import spark.implicits._ 不起作用。要转换为dataframes的底层对象确实有数据。
错误堆栈似乎与使用scala运行时反射的运行时隐式解析有关。
注意，我已经检查过了，spark和编译的代码都使用相同版本的scala（2.11）。
我得到的例外是：

java.lang.RuntimeException: java.util.NoSuchElementException: head of empty list
scala.collection.immutable.Nil$.head(List.scala:420)
scala.collection.immutable.Nil$.head(List.scala:417)
scala.collection.immutable.List.map(List.scala:277)
scala.reflect.internal.Symbols$Symbol.parentSymbols(Symbols.scala:2117)
scala.reflect.internal.SymbolTable.openPackageModule(SymbolTable.scala:301)
scala.reflect.internal.SymbolTable.openPackageModule(SymbolTable.scala:341)
scala.reflect.runtime.SymbolLoaders$LazyPackageType$$anonfun$complete$2.apply$mcV$sp(SymbolLoaders.scala:74)
scala.reflect.runtime.SymbolLoaders$LazyPackageType$$anonfun$complete$2.apply(SymbolLoaders.scala:71)
scala.reflect.runtime.SymbolLoaders$LazyPackageType$$anonfun$complete$2.apply(SymbolLoaders.scala:71)
scala.reflect.internal.SymbolTable.slowButSafeEnteringPhaseNotLaterThan(SymbolTable.scala:263)
scala.reflect.runtime.SymbolLoaders$LazyPackageType.complete(SymbolLoaders.scala:71)
scala.reflect.internal.Symbols$Symbol.info(Symbols.scala:1514)
scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$1.scala$reflect$runtime$SynchronizedSymbols$SynchronizedSymbol$$super$info(SynchronizedSymbols.scala:174)
scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info$1.apply(SynchronizedSymbols.scala:127)
scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anonfun$info$1.apply(SynchronizedSymbols.scala:127)
scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:123)
scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$1.gilSynchronizedIfNotThreadsafe(SynchronizedSymbols.scala:174)
scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$class.info(SynchronizedSymbols.scala:127)
scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$1.info(SynchronizedSymbols.scala:174)
scala.reflect.internal.Types$TypeRef.thisInfo(Types.scala:2194)
scala.reflect.internal.Types$TypeRef.baseClasses(Types.scala:2199)
scala.reflect.internal.tpe.FindMembers$FindMemberBase.<init>(FindMembers.scala:17)
scala.reflect.internal.tpe.FindMembers$FindMember.<init>(FindMembers.scala:219)
scala.reflect.internal.Types$Type.scala$reflect$internal$Types$Type$$findMemberInternal$1(Types.scala:1014)
scala.reflect.internal.Types$Type.findMember(Types.scala:1016)
scala.reflect.internal.Types$Type.memberBasedOnName(Types.scala:631)
scala.reflect.internal.Types$Type.member(Types.scala:600)
scala.reflect.internal.Mirrors$RootsBase.getModuleOrClass(Mirrors.scala:48)
scala.reflect.internal.Mirrors$RootsBase.getModuleOrClass(Mirrors.scala:66)
scala.reflect.internal.Mirrors$RootsBase.staticPackage(Mirrors.scala:204)
scala.reflect.runtime.JavaMirrors$JavaMirror.staticPackage(JavaMirrors.scala:82)
scala.reflect.internal.Mirrors$RootsBase.init(Mirrors.scala:263)
scala.reflect.runtime.JavaMirrors$class.scala$reflect$runtime$JavaMirrors$$createMirror(JavaMirrors.scala:32)
scala.reflect.runtime.JavaMirrors$$anonfun$runtimeMirror$1.apply(JavaMirrors.scala:49)
scala.reflect.runtime.JavaMirrors$$anonfun$runtimeMirror$1.apply(JavaMirrors.scala:47)
scala.reflect.runtime.Gil$class.gilSynchronized(Gil.scala:19)
scala.reflect.runtime.JavaUniverse.gilSynchronized(JavaUniverse.scala:16)
scala.reflect.runtime.JavaMirrors$class.runtimeMirror(JavaMirrors.scala:46)
scala.reflect.runtime.JavaUniverse.runtimeMirror(JavaUniverse.scala:16)
scala.reflect.runtime.JavaUniverse.runtimeMirror(JavaUniverse.scala:16)

我的工作假设是我缺少一个依赖项或导入，这是一种scala主义。
我还没有找到任何其他提到这个问题的地方。最终我认为这可能是由于导入/依赖，但到目前为止，我还不太清楚它是什么。非常感谢您的帮助。我很想知道解决这个问题的方法，或者通过比其他方法更不神奇的方法来创建Dataframe toDf() .
spark信息：

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.3.2.0-mapr-1901
      /_/

Using Scala version 2.11.8 (OpenJDK 64-Bit Server VM, Java 1.8.0_191)
hbasescalaapache-sparkreflectionlivy

1条答案

按热度按时间
ktca8awb

ktca8awb1#

我在同一版本的spark上遇到了这个错误，但是，我在从hdfs读取csv时看到错误出现，这就是我正在做的一个示例：

val csv: DataFrame = ss
  .read
  .option("header", "true")
  .option("mode", "DROPMALFORMED")
  .csv(filePath)
println(csv.count())

这就是我看到错误起源于Spark的地方。
我举了一个失败的小例子，试图找出问题的根源。我使用livyscala编程api将作业提交给spark。我发现它之所以失败是因为我通过livy将类型作为参数传递给spark，这是有意义的，因为这是一个scala反射错误。
例如，此操作失败：

case class FailingJob(someSeq: Seq[String], filePath: String) {
...
def call(scalaJobContext: ScalaJobContext): Unit = {
    // It doesn't really matter what I do here.
// Main this is that the seq is used in some way.
val mappedSeq = someSeq.map(s => s.toUpperCase())

val ss: SparkSession = scalaJobContext.sparkSession

val csv: DataFrame = ss
  .read
  .option("header", "true")
  .option("mode", "DROPMALFORMED")
  .csv(filePath)

println(csv.count())
...
 for {
     _ <- livyClient.submit(FailingJob(someSeq, path).call)
...

鉴于这是成功的：

case class SuccessfulJob(someArray: Array[String], filePath: String) {
...
def call(scalaJobContext: ScalaJobContext): Unit = {
    // It doesn't really matter what I do here.
// Main this is that the seq is used in some way.
val mappedSeq = someArray.map(s => s.toUpperCase())

val ss: SparkSession = scalaJobContext.sparkSession

val csv: DataFrame = ss
  .read
  .option("header", "true")
  .option("mode", "DROPMALFORMED")
  .csv(filePath)

println(csv.count())
...
 for {
     _ <- livyClient.submit(SuccessfulJob(someArray, path).call)
...

如果我传入一个参数 Seq 这失败了，这让我认为这是serrialization/deserialization wihtin的问题 kryo . 另外要注意的是，如果我将值作为对象的属性引用，则不会引发此错误。我尝试升级到spark 2.4，但运气不好。我用的是livy版本 0.6.0-incubating . 我目前的工作是转换要使用的对象 Array 类型而不是 Seq . 我怀疑其他特定于scala的类型也会失败，尽管我还没有尝试过。
这是我对这个问题的复制品，包括解决方法。我很感激这不能回答这个问题，但希望它能帮助其他人在其他场景中调试这个问题。我还向livy提交了一个问题，看看他们是否能对正在发生的事情有更多的了解。

赞(0）分享  回复(0）举报  2021-06-09
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com