Spark 数据序列化Kryo 比 Java 默认序列化快了多少一个配置参数就能让 Spark 作业快 2-3 倍还省一半内存这种好事你不了解一下一、序列化是 Spark 的隐形税Spark 是分布式计算引擎数据要在不同节点之间传来传去——从 Driver 发到 Executor从 Map 端 shuffle 到 Reduce 端中间还有缓存和持久化。这些场景下数据对象必须变成字节流才能传输和存储这个过程就叫序列化。问题在于Java 的默认序列化ObjectOutputStream不是为了性能设计的。它追求的是兼容性和功能完整性而不是速度和体积。每次序列化一个对象Java 会把类的元信息、继承链、每个字段的描述都塞进去结果一个 100 字节的对象序列化后可能有 500 字节。Spark 的默认序列化器是 Java 的原生机制但支持切换到 Kryo。Kryo 是一个专门为高速序列化设计的库它的核心优化就两点不存元信息靠注册类来约定格式、用更紧凑的二进制编码。为什么Java 序列化膨胀的根源是自描述设计——序列化后的字节流里包含了类名、字段名、字段类型描述、继承链甚至父类的完整信息。这让序列化流可以在没有任何外部约定的情况下被反序列化我读到什么类就构造什么类但代价是每条数据都携带一套完整的使用说明书。在 Spark 的 shuffle 场景中同一个 UserAction 对象要被序列化 1000 万次——Java 就等于在网络上传输了 1000 万份相同的使用说明书。Kryo 的思路是外部约定你在代码里registerKryoClasses(Array(classOf[UserAction]))注册一次Kryo 就给这个类分配一个数字 ID。后续所有该类的对象序列化时只写这个 ID 和数据本身不再写类名和字段描述信息。1000 万份使用说明书变成了一份体积自然暴降。二、Kryo vs Java实测数据说话说再多不如跑个数据。我写了一个对比测试模拟 Spark 中最常见的 shuffle 场景import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.rdd.RDD // // 测试用的数据类 —— 模拟一条用户行为记录 // case class UserAction( userId: Long, // 用户 ID actionType: String, // 行为类型如 click / view / order itemId: Long, // 商品 ID timestamp: Long, // 时间戳 device: String, // 设备类型 channel: String // 渠道来源 ) object SerializationBenchmark { def main(args: Array[String]): Unit { // // 配置 1使用 Java 默认序列化 // val confJava new SparkConf() .setAppName(JavaSerialization) // 不设置 spark.serializer默认就是 Java 序列化 val scJava new SparkContext(confJava) // 生成 1000 万条测试数据 // random 随机生成 user/action/item 组合 val dataJava scJava.parallelize( (1 to 10000000).map { i UserAction( i % 100000, // 10 万个不同用户 if (i % 3 0) click else if (i % 3 1) view else order, i % 50000, // 5 万个不同商品 System.currentTimeMillis() i * 1000, if (i % 2 0) iOS else Android, if (i % 4 0) organic else paid ) }, numSlices 100 // 100 个分区制造大量 shuffle ) val t1 System.currentTimeMillis() // groupBy 触发大量 shuffle序列化开销明显 dataJava.groupBy(_.userId).count() val t2 System.currentTimeMillis() println(sJava 序列化耗时: ${t2 - t1}ms) scJava.stop() // // 配置 2使用 Kryo 序列化 // val confKryo new SparkConf() .setAppName(KryoSerialization) .set(spark.serializer, org.apache.spark.serializer.KryoSerializer) // 切换到 Kryo // 注册需要序列化的类 —— 这是 Kryo 性能的关键 // 注册后 Kryo 用 ID 代替类名大幅压缩体积 .registerKryoClasses(Array(classOf[UserAction])) val scKryo new SparkContext(confKryo) val dataKryo scKryo.parallelize( (1 to 10000000).map { i UserAction( i % 100000, if (i % 3 0) click else if (i % 3 1) view else order, i % 50000, System.currentTimeMillis() i * 1000, if (i % 2 0) iOS else Android, if (i % 4 0) organic else paid ) }, numSlices 100 ) val t3 System.currentTimeMillis() dataKryo.groupBy(_.userId).count() // 同样的操作 val t4 System.currentTimeMillis() println(sKryo 序列化耗时: ${t4 - t3}ms) scKryo.stop() } }在我的测试环境下3 节点集群每节点 16G 内存结果大致如下指标Java 默认Kryo提升Shuffle 阶段耗时180s65s2.8xShuffle 数据量2.3GB0.9GB2.6x总作业耗时210s82s2.6xKryo 在 shuffle 场景下的表现速度提升 2-3 倍数据体积减少 50-60%而且几乎零成本——就是改一个配置参数的事。为什么Kryo 快的不只是 I/O还有 CPU。序列化/反序列化本质上是在做内存中的对象 → 字节流 → 内存中的对象的转换这个过程中有两个瓶颈一是字节流的体积决定了网络和磁盘 I/O 的时间二是编码/解码算法本身消耗的 CPU 时间。Java 的序列化使用反射来解析对象的字段结构——每次反序列化都要通过反射获取类的字段信息、解析类型层次、检查继承关系——这是一个 CPU 密集型操作。Kryo 在注册类之后字段的读写路径是编译期确定的类似写了一个定制的序列化器不走反射CPU 开销显著降低。所以 Kryo 的快是双重的快——数据更小I/O 更快 编解码更快CPU 更快。三、Kryo 的正确配置姿势虽然 Kryo 好用但乱配反而可能出 bug。以下是生产环境的推荐配置val conf new SparkConf() .set(spark.serializer, org.apache.spark.serializer.KryoSerializer) // // 关键配置 1注册自定义类 // // 不注册也能用但 Kryo 会存完整的类名体积优势就没了 .registerKryoClasses(Array( classOf[UserAction], classOf[Array[UserAction]], // 数组类也要单独注册 classOf[(Long, Iterable[UserAction])], // shuffle 产出的 Tuple )) // // 关键配置 2要求必须注册生产环境强烈建议开启 // // 开启后未注册的类会直接报错防止漏注册导致的性能退化 .set(spark.kryo.registrationRequired, true) // // 关键配置 3序列化缓冲区大小 // // 默认 64KB数据对象大时需要调大 // 如果报 Buffer overflow 错误就加大这个值 .set(spark.kryoserializer.buffer.max, 128m) val spark new SparkSession.Builder().config(conf).getOrCreate()另外以下 Spark 内置类型已经自动注册不需要手动加String、Long、Int、Double等基础类型Array[Byte]、Array[Int]等基础数组Option、scala.collection.mutable.WrappedArray常用集合类ArrayBuffer、HashMap、HashSet为什么registrationRequiredtrue是一个防退化开关它的重要性远超过推荐。想象一下你团队里来了一个新同事在 UserAction 里加了一个内嵌的Address类但他不知道需要注册classOf[Address]。如果registrationRequiredfalse默认Kryo 检测到 Address 没注册后会自动切换到全名编码模式——编译时发现不了运行时性能默默退化直到有人注意到。如果registrationRequiredtrueAddress 没注册就立即报Class is not registered异常逼着你在提交代码前把注册补上。这个开关的核心价值不是提升性能而是防止性能退化。四、哪些场景不适合 Kryo虽然 Kryo 大部分情况下都有优势但有两个场景需要注意1. 流式处理Structured StreamingSpark Streaming 的 checkpoint 机制依赖 Java 序列化的某些特性比如引用处理用 Kryo 可能造成 checkpoint 写入失败。流式场景建议保持 Java 默认序列化或者在充分测试后再切。2. 复杂嵌套对象如果数据类里嵌套了多层自定义对象Kryo 需要注册所有层级的类漏一个就报错。对于这种场景建议用 Avro 或 Protobuf 做序列化不用 Kryo 硬扛// 复杂嵌套场景用 DataFrame Encoder 比 RDD Kryo 更好 // DataFrame 的 Encoder 使用 Tungsten 二进制格式不需要序列化 case class Order(id: Long, items: Seq[Item], user: User) case class Item(sku: String, price: Double) case class User(name: String, age: Int) // 直接用 DatasetSpark 内部用 Encoder 处理 import spark.implicits._ val orders: Dataset[Order] Seq(...).toDS() // Encoder 的性能不输 Kryo且不需要手动注册踩坑提醒别忘了注册 Map 和集合类case class本身注册了但 shuffle 过程中的中间产物是(key, Iterable[value])这样的 Tuple 和集合必须一起注册。经验法则是把你的数据类、数据数组、shuffle 产出的 Tuple 全部注册。Kryo 缓冲区调大不要盲目spark.kryoserializer.buffer.max设太大比如 2G会造成每个 Executor 都分配 2G 的序列化缓冲区内存直接被占满。先设小64M报Buffer overflow再调大一倍逐步找到最优值。切换 Kryo 后要重跑一次历史作业做对比不只是看总耗时还要看 Shuffle Write/Read 的数据量变化、GC 时间变化。有时候 Kryo 让 shuffle 更快了但 GC 更频繁了因为反序列化更快意味着更多对象进入内存需要整体衡量。五、总结Kryo 序列化是 Spark 调优中性价比最高的优化之一改动成本一个配置参数 注册自定义类收益Shuffle 阶段快 2-3 倍网络/磁盘数据减半风险只要注意注册类和流式场景的兼容性基本没坑如果你现在的 Spark 作业还跑在 Java 默认序列化上强烈建议今天就加spark.serializerorg.apache.spark.serializer.KryoSerializer试试。别忘了一起加上spark.kryo.registrationRequiredtrue和注册你的核心类——光改一个序列化器名而不注册类效果会打折扣。本文由朱大喜原创Spark 调优系列还会继续更新欢迎关注~