序列化在分布式系統(tǒng)中扮演著重要的角色，優(yōu)化Spark程序時(shí)，首當(dāng)其沖的就是對(duì)序列化方式的優(yōu)化。Spark為使用者提供兩種序列化方式：
Java Serialization: 默認(rèn)的序列化方式。
Kryo Serialization: 相較于 Java Serialization 的方式，速度更快，空間占用更小，但并不支持所有的序列化格式，同時(shí)使用的時(shí)候需要注冊(cè)class。spark-sql中默認(rèn)使用的是kyro的序列化方式。
下文將會(huì)講解kryo的使用方式并對(duì)比性能。

配置

可以在spark-default.conf設(shè)置全局參數(shù)，也可以代碼中初始化時(shí)對(duì)SparkConf設(shè)置 conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") ，該參數(shù)會(huì)同時(shí)作用于機(jī)器之間數(shù)據(jù)的shuffle操作以及序列化rdd到磁盤(pán)，內(nèi)存。

Spark不將Kyro設(shè)置成默認(rèn)的序列化方式是因?yàn)樗枰獙?duì)類進(jìn)行注冊(cè)，官方強(qiáng)烈建議在一些網(wǎng)絡(luò)數(shù)據(jù)傳輸很大的應(yīng)用中使用kyro序列化。

val conf = new SparkConf()
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
conf.registerKryoClasses(Array(classOf[MyClass1],classOf[MyClass2]))
val sc = new SparkContext(conf)

如果你要序列化的對(duì)象比較大，可以增加參數(shù)spark.kryoserializer.buffer所設(shè)置的值。

如果你沒(méi)有注冊(cè)需要序列化的class，Kyro依然可以照常工作，但會(huì)存儲(chǔ)每個(gè)對(duì)象的全類名(full class name)，這樣的使用方式往往比默認(rèn)的 Java serialization 還要浪費(fèi)更多的空間。

可以設(shè)置 spark.kryo.registrationRequired 參數(shù)為 true，使用kyro時(shí)如果在應(yīng)用中有類沒(méi)有進(jìn)行注冊(cè)則會(huì)報(bào)錯(cuò)：

這里寫(xiě)圖片描述

如上這個(gè)錯(cuò)誤需要添加

sparkConf.registerKryoClasses(
    Array(classOf[scala.collection.mutable.WrappedArray.ofRef[_]],
    classOf[MyClass]))

下面的 demo 將會(huì)演示不同方式的序列化對(duì)空間占用的情況。

DEMO

case class Info(name: String ,age: Int,gender: String,addr: String)

object KyroTest {
  def main(args: Array[String]) {
  
  val conf = new SparkConf().setMaster("local[2]").setAppName("KyroTest")
      conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      conf.registerKryoClasses(Array(classOf[Info]))
  val sc = new SparkContext(conf)

  val arr = new ArrayBuffer[Info]()

  val nameArr = Array[String]("lsw","yyy","lss")
  val genderArr = Array[String]("male","female")
  val addressArr = Array[String]("beijing","shanghai","shengzhen","wenzhou","hangzhou")

  for(i <- 1 to 1000000){
    val name = nameArr(Random.nextInt(3))
    val age = Random.nextInt(100)
    val gender = genderArr(Random.nextInt(2))
    val address = addressArr(Random.nextInt(5))
    arr.+=(Info(name,age,gender,address))
    }

  val rdd = sc.parallelize(arr)

  //序列化的方式將rdd存到內(nèi)存
  rdd.persist(StorageLevel.MEMORY_ONLY_SER)
  rdd.count()
  }
}

結(jié)果

可以在web ui中看到緩存的rdd大?。?/p>

這里寫(xiě)圖片描述