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

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

Serializable

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

kryo

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

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

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

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)

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.ORCUtilclass 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 ORCUtilval 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)}
}