如何在Spark临时视图中使用递归Case Class并通过SQL操作其字段

我有一个递归Case Class：

case class Dummy(id: Int, children: List[Dummy])

Spark默认编码器不支持循环引用，因此我采用了支持递归的Kryo自定义编码器：

val myEncoder: Encoder[Dummy] = Encoders.kryo[Dummy]
val ds = spark.createDataset(Seq(Dummy(1, null)))(myEncoder)
ds.createOrReplaceTempView("myTable")

该方案能正常操作Dataset，但执行Spark SQL查询时出现错误：

select id, size(children) as childCount from myTable

错误信息：

org.apache.spark.sql.AnalysisException: cannot resolve '`id`' given input columns: [myTable.value]; line 1 pos 7;

查看表结构发现Schema为：

value: binary (nullable = true)

注：此问题与spark-using-recursive-case-class类似，但需求不同——我需要通过SQL而非Dataset API操作Case Class字段，已尝试Kryo方案无效，请问Spark是否支持在SQL中自动反序列化Case Class？

Spark不支持直接通过SQL自动反序列化Kryo编码的二进制对象。因为Encoders.kryo会把整个Case Class序列化为单个二进制字段（即value: binary），Spark SQL无法识别该二进制字段内部的结构，自然无法直接引用id、children这类嵌套字段。

要实现SQL操作递归Case Class的字段，可采用以下两种方案：

方案一：自定义UDF反序列化二进制字段 #

编写UDF将二进制的value字段反序列化为Dummy对象，再提取对应字段。示例代码：

import org.apache.spark.sql.functions.udf
import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.Input

// Kryo实例非线程安全，生产环境建议用ThreadLocal管理
val kryoLocal = new ThreadLocal[Kryo] {
  override def initialValue(): Kryo = {
    val kryo = new Kryo()
    kryo.register(classOf[Dummy])
    kryo.register(classOf[scala.collection.immutable.Nil$])
    kryo.register(classOf[scala.collection.immutable.$colon$colon[_]])
    kryo
  }
}

// 提取id字段的UDF
val extractId = udf((binary: Array[Byte]) => {
  val input = new Input(binary)
  val dummy = kryoLocal.get().readObject(input, classOf[Dummy])
  input.close()
  dummy.id
})

// 提取children长度的UDF
val extractChildrenSize = udf((binary: Array[Byte]) => {
  val input = new Input(binary)
  val dummy = kryoLocal.get().readObject(input, classOf[Dummy])
  input.close()
  Option(dummy.children).map(_.size).getOrElse(0)
})

// 注册UDF到Spark
spark.udf.register("extractId", extractId)
spark.udf.register("extractChildrenSize", extractChildrenSize)

注册完成后即可在SQL中使用：

select extractId(value) as id, extractChildrenSize(value) as childCount from myTable

方案二：改用Spark SQL原生支持的递归数据类型 #

手动定义递归Schema，将Case Class转换为Spark SQL原生支持的嵌套结构，无需依赖Kryo编码器。示例代码：

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

// 递归定义Schema
val dummySchema: StructType = StructType(Seq(
  StructField("id", IntegerType, nullable = false),
  StructField("children", ArrayType(dummySchema), nullable = true)
))

// 将Dummy对象转换为Row格式
def dummyToRow(d: Dummy): Row = {
  val childrenRows = Option(d.children).map(_.map(dummyToRow)).getOrElse(Seq.empty)
  Row(d.id, childrenRows)
}

// 构建DataFrame并注册临时视图
val rows = Seq(Dummy(1, null)).map(dummyToRow)
val df = spark.createDataFrame(spark.sparkContext.parallelize(rows), dummySchema)
df.createOrReplaceTempView("myTable")

此时可直接执行SQL查询：

select id, size(children) as childCount from myTable

该方案是Spark原生支持的，SQL可直接识别所有字段，是更推荐的解决方案。

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