为什么Spark Dataset丢失全部schema仅返回byte[]？如何修复该问题？

• 核心原因是你手动指定了Encoders.kryo[HeatSensorEvent]作为样例类的编码器：Encoders.kryo[T]属于序列化专用编码器，作用是将整个T类型的对象直接序列化为单个二进制字节数组，Spark SQL不会解析对象内部的字段结构，所以Dataset的Schema只会保留一个名为value的binary列，自然看不到结构化字段。
• 额外说明：你在SparkConf中配置的Kryo序列化属于Spark作业层面的序列化配置，仅用于shuffle、节点间数据传输、RDD缓存时的序列化，和Spark SQL/Dataset层面的Encoder没有关系，不影响Schema解析。
• 次要原因是Scala原生的Enumeration类型，Spark默认的产品编码器（即隐式自动生成的case class编码器）无法直接识别，所以你才会尝试手动指定Kryo编码器，反而导致了Schema丢失。

方案1：修改枚举实现为Spark原生支持的形式（推荐） #

将原有的Enumeration枚举改为sealed trait + case object的实现方式，Spark可以自动识别该类型的结构，无需手动指定编码器：

object Model {
  type Timestamp = Long
  sealed trait EventType
  case object TEMPERATURE_CHANGE extends EventType

  case class HeatSensorEvent(
                              eventId: String,
                              sensorId: String,
                              deviceId: String,
                              eventType: EventType,
                              timestamp: Timestamp,
                              temperature: Double
                            )
}

然后删除主函数中你手动定义的两个implicit Kryo编码器即可：

def main(args: Array[String]): Unit = {
    import Model._ // 导入样例类和枚举，Spark会自动生成对应的编码器
    val heatSensorEventDs: Dataset[HeatSensorEvent] = spark
      .createDataset(heatSensorEventData).as[HeatSensorEvent]
    heatSensorEventDs.show
    heatSensorEventDs.printSchema()
}

方案2：保留原有Enumeration，自定义编码器 #

如果不想修改原有枚举定义，可以自定义EventType的编码器，不要使用Kryo编码器：

implicit val eventTypeEncoder: Encoder[Model.EventType.EventType] = Encoders.javaSerialization[Model.EventType.EventType]

同样删除手动定义的HeatSensorEvent的Kryo编码器，Spark会自动结合自定义的EventType编码器生成HeatSensorEvent的结构化编码器，保留完整Schema。

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