Spark 4.x中如何向UDF传递结构体数组参数？代码报错排查

问题原因 #

Spark 4.x重构了编码器系统，引入Agnostic Encoders以提升跨语言、多数据源的类型适配能力，但这导致原Spark 3.x中依赖隐式Row编码器的UDF场景不再兼容。当UDF参数声明为Seq[Row]时，Spark无法为UnboundRowEncoder匹配到正确的序列化逻辑，从而抛出MatchError。

解决方案 #

方案1：直接使用Case Class作为UDF参数（推荐） #

利用已定义的Case Class类型替代Row，Spark可以自动推导正确的编码器，避免编码适配问题：

// 保持原Case Class定义
case class N(navs: Int)
case class I(first: String, second: Seq[N])
val h = Seq(I("a", Seq(N(10), N(15)))).toDF

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

// UDF参数改为Seq[N]，直接操作Case Class字段
val reduceItems = (items: Seq[N]) => {
  items.map(_.navs).reduce(_ + _)
}

val reduceItemsUdf = udf(reduceItems)

h.select(reduceItemsUdf($"second").as("r")).show()

方案2：显式指定UDF的输入编码器（兼容Row场景） #

如果业务必须使用Row，可以显式定义输入结构的Schema并指定编码器：

case class N(navs: Int)
case class I(first: String, second: Seq[N])
val h = Seq(I("a", Seq(N(10), N(15)))).toDF

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
import org.apache.spark.sql.expressions.UserDefinedFunction

// 定义结构体数组的Schema
val nSchema = StructType(Seq(StructField("navs", IntegerType)))
val arrayNSchema = ArrayType(nSchema)

// 显式指定输入和输出类型创建UDF
val reduceItems = (items: Seq[Row]) => {
  items.map(_.getAs[Int]("navs")).reduce(_ + _)
}

val reduceItemsUdf = udf(reduceItems, IntegerType, arrayNSchema)

h.select(reduceItemsUdf($"second").as("r")).show()

方案3：使用强类型Dataset API替代UDF #

通过Dataset的强类型操作绕过UDF的编码问题，更符合Spark 4.x的设计方向：

case class N(navs: Int)
case class I(first: String, second: Seq[N])
val ds = Seq(I("a", Seq(N(10), N(15)))).toDS()

// 直接通过map操作处理数据
val resultDs = ds.map { item =>
  val sum = item.second.map(_.navs).reduce(_ + _)
  (item.first, sum)
}.toDF("first", "r")

resultDs.show()

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