Scala3项目中Spark Case Class Schema与UDF的TypeTag缺失问题修复

问题现象 #

以下代码在Spark Shell REPL中可正常运行，但在Scala 3.3.0项目构建时抛出No TypeTag available错误：

示例1

import org.apache.spark.sql.Encoders
import org.apache.spark
import scala.reflect.runtime.universe._

case class StationX(  stnId: Int,
                    wbanId: Int,
                    lat: Double,
                    lon: Double)
object Inst {
  def getSchema[T <: Product : TypeTag] = Encoders.product[T].schema
  val ss = getSchema[StationX]
}

错误信息：No TypeTag available for observatory.StationX（指向val ss = getSchema[StationX]）

示例2

添加以下对象后同样出现TypeTag缺失错误：

package observatory
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.Encoders
import scala.reflect.runtime.universe._

case class CC(i: Int)
object SparkInstance {
  val spark = SparkSession
    .builder()
    .appName("Spark SQL UDF scalar example")
    .getOrCreate()

  def getSchema[T <: Product : TypeTag] = Encoders.product[T].schema

  // No TypeTag available for Double
  val random = udf(() => Math.random())

  // No TypeTag available for Int
  val plusOne = udf((x: Int) => x + 1)

  // No TypeTag available for observatory.CC
  val ss = getSchema[CC]
}

解决方案 #

1. 添加Scala反射依赖到build.sbt

Scala 3默认不包含反射库，需要显式引入：

libraryDependencies += "org.scala-lang" % "scala-reflect" % scalaVersion.value

2. 延迟求值避免初始化时隐式缺失

对象中的val会在类初始化时立即求值，此时Scala 3的隐式TypeTag可能还未生成。将val改为def（惰性求值，调用时才推导隐式）即可解决：

// 原代码
val ss = getSchema[StationX]
// 修改后
def ss = getSchema[StationX]

3. UDF改用显式Encoder替代TypeTag

Spark 3.3+提供了无需TypeTag的UDF重载方法，显式指定Encoder即可绕过TypeTag依赖：

// 原无参数UDF
val random = udf(() => Math.random())
// 修改后
val random = udf(() => Math.random(), Encoders.scalaDouble)

// 原单参数UDF
val plusOne = udf((x: Int) => x + 1)
// 修改后
val plusOne = udf((x: Int) => x + 1, Encoders.scalaInt)

4. 升级Spark到3.3+版本

Spark 3.2对Scala 3的支持不完善，3.3及以上版本优化了Scala 3的兼容性，确保项目使用Spark 3.3+并移除build.sbt中对Spark 3.2的排除项。

修正后的完整代码 #

示例1修正版

import org.apache.spark.sql.Encoders
import scala.reflect.runtime.universe._

case class StationX(stnId: Int, wbanId: Int, lat: Double, lon: Double)
object Inst {
  def getSchema[T <: Product : TypeTag] = Encoders.product[T].schema
  // 改用def延迟求值
  def ss = getSchema[StationX]
}

示例2修正版

package observatory
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.Encoders
import scala.reflect.runtime.universe._

case class CC(i: Int)
object SparkInstance {
  val spark = SparkSession
    .builder()
    .appName("Spark SQL UDF scalar example")
    .getOrCreate()

  def getSchema[T <: Product : TypeTag] = Encoders.product[T].schema

  // 显式指定Encoder
  val random = udf(() => Math.random(), Encoders.scalaDouble)
  val plusOne = udf((x: Int) => x + 1, Encoders.scalaInt)

  // 改用def延迟求值
  def ss = getSchema[CC]
}

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