Kotlin下Spark从嵌套POJO创建DataFrame报Schema不支持错误如何解决

首先明确：Spark完全支持从嵌套Kotlin数据类集合创建DataFrame #

你遇到的报错不是Spark不支持嵌套结构，而是选用的DataFrame构造方式存在适配问题，核心原因有两个：

• spark.createDataFrame(rdd1, PersonaMetaData::class.java) 这个API依赖Java Bean内省机制推断Schema，Kotlin嵌套数据类的泛型集合（List<Persona>、List<Persona_Items>）在运行时存在泛型擦除，Spark的内省逻辑无法识别嵌套自定义类型的结构，所以抛出类型不支持的错误。
• 你做了多余的转换步骤：已经通过toDS生成了类型安全的Dataset，完全不需要转成RDD再构造DataFrame，绕路反而触发了API的适配问题。

修复方案 #

最简单的修复方式是直接从Dataset转DataFrame，Spark原生内置的Kotlin编码器可以正确识别嵌套数据类的Schema，不需要手动处理。
将原来转RDD再构造DF的两行代码，替换为下面这行即可：

val df = spark.toDS(listOf(personMetaData)).toDF()

如果你的Spark版本低于3.0，需要额外导入Kotlin编码器依赖的包：

import org.apache.spark.sql.kotlin.encoders.*

调整后运行df.show(false)可以正常打印嵌套结构，嵌套的listPersona、listPersonaItems会被自动识别为Array类型的嵌套Struct结构。

可选备用方案 #

如果你确实需要从RDD构造DataFrame，可以手动定义Schema，将RDD映射为RDD<Row>后再构造，代码示例如下，这种方式灵活性更高但代码更冗余，非特殊场景不推荐使用：

// 手动定义全量嵌套Schema
val itemSchema = DataTypes.createStructType(arrayOf(
    DataTypes.createStructField("key1", DataTypes.IntegerType, true),
    DataTypes.createStructField("key2", DataTypes.StringType, true)
))
val personaSchema = DataTypes.createStructType(arrayOf(
    DataTypes.createStructField("persona_type", DataTypes.StringType, true),
    DataTypes.createStructField("created_using_algo", DataTypes.StringType, true),
    DataTypes.createStructField("version_algo", DataTypes.StringType, true),
    DataTypes.createStructField("createdAt", DataTypes.LongType, true),
    DataTypes.createStructField("listPersonaItems", DataTypes.createArrayType(itemSchema), true)
))
val metaSchema = DataTypes.createStructType(arrayOf(
    DataTypes.createStructField("user_id", DataTypes.IntegerType, true),
    DataTypes.createStructField("persona_created", DataTypes.BooleanType, true),
    DataTypes.createStructField("persona_createdAt", DataTypes.LongType, true),
    DataTypes.createStructField("listPersona", DataTypes.createArrayType(personaSchema), true)
))

// 将实体RDD映射为Row RDD再构造DF
val rowRdd = spark.toDS(listOf(personMetaData)).rdd().map { meta ->
    val personaRows = meta.listPersona.map { p ->
        val itemRows = p.listPersonaItems.map { Row(it.key1, it.key2) }
        Row(p.persona_type, p.created_using_algo, p.version_algo, p.createdAt, itemRows)
    }
    Row(meta.user_id, meta.persona_created, meta.persona_createdAt, personaRows)
}
val df = spark.createDataFrame(rowRdd, metaSchema)

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