Problem
序列化一个Scala对象为文本或者二进制数据，以便支持持久化或者网络传输等需求，并通过读取这些数据，可以反序列化出这个对象
Feature requests
- easy to use
- 支持自定义序列化(如部分member)
- 尽可能的编译期检查
- Schema Evolution?
Key metrics
- 序列化/反序列化 速度
- 序列化数据 空间占用
- 通用性
Solution
extend or mixin Serializable trait
scala Serializable 实际上就是一个java.io.Serializable的universal trait
package scala
/**
* Classes extending this trait are serializable across platforms (Java, .NET).
*/
trait Serializable extends Any with java.io.Serializable
@SerialVersionUID
@SerialVersionUID(1000L)
class Foo extends Serializable {
// class code here
}
Static annotation SerialVersionUID可以与Serialization一起使用
If no serialVersionUID is declared, JVM will use its own algorithm to generate a default SerialVersionUID.
When to specify SerialVersionUID?
SerialVersionUID的目的是为了检查序列化和反序列化的类是否兼容。
- 第一种情况，序列化的目的只是为了在网络上即时传输，如rpc, mq等，或者在实现上考虑，为了节省内存，只保存序列化之后的对象，如spark cache等，一般来说可以不显式指定SerialVersionUID
- 第二种情况，需要直接使用序列化来持久化对象，如将训练好的模型存储到文件系统上，就最好指定SerialVersionUID，且当类不向前兼容的时候，应该重新生成SerialVersionUID
- 第三种情况，如果对各编译器，不同的JVM 兼容性要求很高，如
Java(TM) Object Serialization Specification建议，应当给每个序列化类都指定SerialVersionUID
按需序列化一部分对象?
当一个类mixin Serializable 之后，整个类的实例(all members)都会被序列化，但有时候这并不是我们需要的
- 对象持有非常大的member，序列化和反序列化的开销很大，而我们并不需要序列化它
- 业务逻辑中，大量序列化和反序列化的开销成为瓶颈，需要优化
- 对象member不可/难以序列化，如网络连接，数据流等，或者是引用第三方库中不可序列化的对象
Solution 1: hygienic closure
- 通过closure来避免序列化整个实例, 而是根据需要传参
def closureFunction[E,D,R](enclosed: E)(gen: E => (D => R)) = gen(enclosed)
class Foo {
val v1 = 42
val v2 = 73
val n = new NotSerializable
// use shim function to enclose *only* the values of 'v1' and 'v2'
def f() = closureFunction((v1, v2)) { enclosed =>
val (v1, v2) = enclosed
(x: Int) => (v1 + v2) * x // Desired function, with 'v1' and 'v2' enclosed
}
}
new Foo.f
- auto nulling via closure cleaning 闭包清理
由于 spark 大量使用closure serialization, 当一个closure 包含了一些在闭包函数中不必要的引用时(Scala issue: SI-1419, fixed in 2.12)，就会浪费网络传输带宽，CPU 开销，还有可能引入一些不可被序列化的对象，导致整个闭包无法序列化。
spark 中使用 ClousureCleaner 在运行时遍历对象，可以更精确的排除不必要的引用。
Solution 2: @transient lazy
Static annotation @transient 表示修饰的 member 不需要被序列化
比如一个 SparkJob base class
class SparkJob(args: Args) extends Serializable {
@transient
protected lazy val sparkConf = new SparkConf()
@transient
protected lazy val sc = new SparkContext(sparkConf)
...
}
由于引用的sparkConf, sparkContext都是不可序列化的，
且都不需要被传送到executor上运行，因此可以用@transient表示该成员不需要被序列化
再如，一个算法模型类，需要读取模型文件，并且需要频繁通过ByteBuffer来操作二进制数据
但是ByteBuffer本身又是不可序列化的, 此时可以使用 @transient (private) lazy pattern
其中 @trainsient 可以避免 overhead，lazy 可以第一次被调用时正确地初始化以避免NPE
class Model(
val model: Array[Byte],
val offset: Array[Byte]
) extends Serializable {
@transient private lazy val offsetBuffer = ByteBuffer.wrap(offset).order(ByteOrder.LITTLE_ENDIAN)
@transient private lazy val modelBuffer = ByteBuffer.wrap(model).order(ByteOrder.LITTLE_ENDIAN)
...
}
这种模式也适用于其他难以被序列化的 member，比如数据库连接，IO stream 等，每个实例只需序列化可以用来重建这些 member 的元信息即可。
Spark 序列化与性能
在如 Spark 这样的分布式计算框架中, broadcast, shuffle, action等操作都会使得对象被序列化。使每个被闭包捕获的变量都可序列化，可以避免异常，但是变量非常大时，容易影响性能，以及有可能造成内存泄露。
Solution 1: Kryo(chill)
在 Spark 中使用 kryo serializer 来获得更小的序列化开销
val conf = new SparkConf().setMaster(...).setAppName(...)
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))
val sc = new SparkContext(conf)
Solution 2: Broadcast
在 Spark 中如果通过闭包引用了一个huge object, 那么这个object会被至少序列化 numPartitions 次，而如果使用broadcast variables, 那么只会被序列化 numNodes 次，通常 numPartitions > numNodes
其他序列化实现
- Protobuf
pros- IDL and languages support
- stable and trusted
cons
- Thrift
pros- more languages support
- rpc service framework
cons
- Avro
- Boopickle
- Pickling
- Scodec
tbc.
Paradigm shift
use more
- function
- typeclass
- case class
- implicit context
References
- http://erikerlandson.github.io/blog/2015/03/31/hygienic-closures-for-scala-function-serialization
- https://github.com/samthebest/dump/blob/master/sams-scala-tutorial/serialization-exceptions-and-memory-leaks-no-ws.md
- http://stackoverflow.com/questions/285793/what-is-a-serialversionuid-and-why-should-i-use-it
