Spark UDF运行时崩溃及单元测试报错问题求助

问题场景 #

编写了Spark UDF工具类及对应的单元测试，运行测试时抛出方法不存在错误，打包Jar部署到Databricks时出现类初始化失败的NoClassDefFoundError，但代码片段在spark-shell或Databricks Notebook中单独运行正常。

1. UDF工具类代码 #

package com.whatever.etl

import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions.udf

object UdfUtils extends Serializable {

  def getKilometersImplx(binCode: Long): Float = {
    import scala.util.Try
    Try((binCode & 0x7fffff).toFloat).toOption.getOrElse(-1f)
  }
  val getKilometers: UserDefinedFunction = udf(getKilometersImplx _) // <<<< line 14

//  val getKilometers: UserDefinedFunction = udf {(binCode: Long) =>
//    import scala.util.Try
//    Try((binCode & 0x7fffff).toFloat).toOption.getOrElse(-1f)
//  }

}

2. 单元测试代码 #

package com.whatever.test

import com.whatever.etl.UdfUtils
import org.scalatest.funsuite.AnyFunSuiteLike

class UDFTest extends AnyFunSuiteLike {
  test("test getKilometers Implementation") {
    val tc: Seq[(Long, Float)] = Seq((0x800000L, 0.0f), (0xffffffL, 0x7fffff.toFloat), (0x800001L, 1.0f))
    tc.foreach { case (binCode, res) => assert(UdfUtils.getKilometersImplx(binCode) == res) }
  }
}

3. 测试运行错误栈 #

An exception or error caused a run to abort: 'org.apache.spark.sql.catalyst.encoders.AgnosticEncoder org.apache.spark.sql.catalyst.ScalaReflection$.encoderForWithRowEncoderSupport(scala.reflect.api.TypeTags$TypeTag)' 
java.lang.NoSuchMethodError: 'org.apache.spark.sql.catalyst.encoders.AgnosticEncoder org.apache.spark.sql.catalyst.ScalaReflection$.encoderForWithRowEncoderSupport(scala.reflect.api.TypeTags$TypeTag)'
    at org.apache.spark.sql.expressions.ScalarUserDefinedFunction$.$anonfun$apply$4(UserDefinedFunction.scala:156)
    at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)
    at scala.collection.IndexedSeqOptimized.foreach(IndexedSeqOptimized.scala:36)
    at scala.collection.IndexedSeqOptimized.foreach$(IndexedSeqOptimized.scala:33)
    at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:38)
    at scala.collection.TraversableLike.map(TraversableLike.scala:286)
    at scala.collection.TraversableLike.map$(TraversableLike.scala:279)
    at scala.collection.AbstractTraversable.map(Traversable.scala:108)
    at org.apache.spark.sql.expressions.ScalarUserDefinedFunction$.apply(UserDefinedFunction.scala:156)
    at org.apache.spark.sql.functions$.udf(functions.scala:7690)
    at com.whatever.etl.UdfUtils$.<init>(UdfUtils.scala:14)

4. 依赖配置（SBT） #

val sparkVersion = "3.3.0"
val scalatestVersion = "3.2.14"
"org.apache.spark" %% "spark-streaming" % sparkVersion % "provided",
"org.apache.spark" %% "spark-core" % sparkVersion % "provided",
"org.apache.spark" %% "spark-sql" % sparkVersion % "provided",
"org.scalatest" %% "scalatest" % scalatestVersion % "test",

5. 额外现象 #

• 单独在spark-shell或Databricks Notebook中运行代码片段可正常执行
• 打包为Jar部署到Databricks时，报错Caused by: NoClassDefFoundError: Could not initialize class com.stla.whatever.etl.xxxxx$

解决思路 #

• 版本兼容性检查：NoSuchMethodError核心原因是编译依赖版本与运行环境版本不匹配。确认Databricks集群的Spark版本是否为3.3.0，若集群版本不同，会导致Spark内部方法签名不一致（比如encoderForWithRowEncoderSupport方法在不同Spark版本中存在差异），需将本地编译的Spark版本调整为与集群一致。
• 延迟UDF初始化：当前代码在单例对象初始化阶段直接创建UDF实例，会触发Spark反射类加载逻辑，而Databricks的类加载环境与本地不同，容易导致单例初始化失败。将UDF改为延迟加载或通过方法返回：

  object UdfUtils extends Serializable {
    def getKilometersImplx(binCode: Long): Float = {
      import scala.util.Try
      Try((binCode & 0x7fffff).toFloat).toOption.getOrElse(-1f)
    }
  
    // 方式1：延迟加载
    lazy val getKilometers: UserDefinedFunction = udf(getKilometersImplx _)
    // 方式2：用方法返回
    // def getKilometers: UserDefinedFunction = udf(getKilometersImplx _)
  }
  

• 规范打包流程：确保打包时不将Spark依赖打入Jar（依赖已标记为provided），避免Jar中的Spark类与集群的Spark类冲突。使用sbt package生成轻量Jar，或用sbt assembly时配置排除provided依赖。
• 优化单元测试：单元测试只需验证业务逻辑方法（如getKilometersImplx），无需初始化UDF实例，避免测试环境中Spark反射相关的类加载问题。

内容的提问来源于stack exchange，提问作者codeaperature