Spark的序列化

对象的序列化:		对象转换为字节序列的过程.
对象的反序列化:	字节序列恢复为对象的过程.
通俗地说序列化就是把内存(jvm)中一个对象的状态通过网络进行传输或者保存到磁盘上,反序列化与之相反.
1
2
3

spark中的序列化:
在spark2.0+版本的官方文档中提到:spark默认提供了两个序列化库,Java自身的序列化和Kryo序列化.
官网的解释是:java序列化灵活,但是速度缓慢;Kryo序列化速度更快且更紧凑,但是支持的类型较少.
1
2
3

Serializable

当使用Serializable方案的时候,你的对象必须继承Serializable接口,类中的属性如果有实例,那也必须是继承Serializable是可序列化的.
如果实在无法序列化那就用transient修饰的;
java代码是在修饰属性的最前面使用transient(短暂的)关键字,scala代码是在修饰属性的最前面使用@transient.
transient的作用是在Serializable序列化的时候,声明某个属性不参与序列化,带来的问题就是如果声明了属性不参与序列化,那这个属性存储的数据也带不过去了.
1
2
3
4

kryo

代码中设置conf.set("spark.serializer",classOf[KryoSerializer].getName)时,这时spark的全局序列化工具就变成了KryoSerializer而不是默认的Serializable方案了.
在生产过程中使用这个设置更简便明了,不需要进行那么多的注册。
当不设置conf.set("spark.kryo.registrationRequired","true")时,spark的所有对象都默认使用KryoSerializer序列化,设置的时候需要对要进行序列化的对象进行注册.
1
2
3

conf.set("spark.serializer", classOf[KryoSerializer].getName)
1

当设置了conf.set("spark.kryo.registrationRequired","true")时,spark中想被KryoSerializer序列化的对象必须得进行注册;注册方法如下:
1

val classes: Array[Class[_]] = Array[Class[_]](classOf[ORCUtil],classOf[StructObjectInspector],classOf[OrcStruct])
conf.set("spark.serializer", classOf[KryoSerializer].getName)
conf.set("spark.kryo.registrationRequired","true")
conf.registerKryoClasses(classes)
1
2
3
4

package RDD
import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.serializers.DefaultArraySerializers.IntArraySerializer
import org.apache.hadoop.hive.ql.io.orc.OrcStruct
import org.apache.hadoop.io.Text
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.serializer.{KryoRegistrator, KryoSerializer}
import org.apache.spark.{SparkConf, SparkContext}
import util.ORCUtil

class MyRegistrator extends KryoRegistrator {
  override def registerClasses(kryo: Kryo): Unit = {
  // 注册序列化类
    kryo.register(Class.forName("scala.reflect.ClassTag$$anon$1"))
    kryo.register(Class.forName("java.lang.Class"))
    kryo.register(Class.forName("scala.collection.mutable.WrappedArray$ofInt"))
    kryo.register(classOf[Array[Int]],new IntArraySerializer)
  }
}
class MapJoinKryo
object MapJoinKryo {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("MapJoin").setMaster("local[*]")
    val classes: Array[Class[_]] = Array[Class[_]](classOf[ORCUtil],classOf[OrcStruct],classOf[Text])
    conf.set("spark.serializer", classOf[KryoSerializer].getName) //设置kryo为序列化方式
    conf.set("spark.kryo.registrationRequired","true")
    conf.set("spark.kryo.registrator","RDD.MyRegistrator")
    conf.registerKryoClasses(classes)

    val sc = new SparkContext(conf)
    val orcUtil = new ORCUtil
    val broadCast: Broadcast[ORCUtil] = sc.broadcast(orcUtil)
    val rdd: RDD[Int] = sc.parallelize(List(1,2))
    val value: RDD[(ORCUtil,Iterable[String])] = rdd.map(f => {
      val orcUtil = broadCast.value
      (orcUtil,"")
    }).groupByKey()
    value.take(200).map(println)
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41