Scala Serialization

作者: DBJ | 来源:发表于2017-01-03 18:12 被阅读0次

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

网友评论

本文标题：Scala Serialization

本文链接：https://www.haomeiwen.com/subject/qlygvttx.html

延伸阅读

深度阅读

您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！