Spark序列化简介

╰+哭是因爲堅強的太久メ 2022-11-07 11:56 203阅读 0赞

参考文章：[Spark序列化][Spark]

[spark之kryo 序列化][spark_kryo]

[Spark序列化入门][Spark 1]

# 1. 什么是序列化和序列化? #

>  *  **序列化是什么**
> 
> 1.  序列化的作用就是可以将对象的内容变成二进制, 存入文件中保存
> 2.  反序列化指的是将保存下来的二进制对象数据恢复成对象
> 
>  *  **序列化对对象的要求**
> 
> 1.  对象必须实现 Serializable 接口
> 2.  对象中的所有属性必须都要可以被序列化, 如果出现无法被序列化的属性, 则序列化失败
> 
>  *  **限制**
> 
> 1.  对象被序列化后, 生成的二进制文件中, 包含了很多环境信息, 如对象头, 对象中的属性字段等, 所以内容相对较大
> 2.  因为数据量大, 所以序列化和反序列化的过程比较慢
> 
>  *  **序列化的应用场景**
> 
> 1.  持久化对象数据
> 2.  网络中不能传输 Java 对象, 只能将其序列化后传输二进制数据

# 2. Java序列化 #

**Java序列化算法**

>  *  Serialization（序列化）是一种将对象以一连串的字节描述的过程；
>  *  反序列化deserialization是一种将这些字节重建成一个对象的过程。
>  *  Java序列化API提供一种处理对象序列化的标准机制。

**为什么要进行序列化？**

> Java中，一切都是对象，在分布式环境中经常需要将Object从这一端网络或设备传递到另一端。  
> 这就需要有一种可以在两端传输数据的协议。  
> Java序列化机制就是为了解决这个问题而产生。

**如何进行序列化？**

> 一个对象能够序列化的前提是实现Serializable接口，Serializable接口没有方法，更像是个标记。  
> 有了这个标记的Class就能被序列化机制处理。

**序列化过程:**

a、实现Serializable接口

![81899018d926236430971b75fbbf802e.png][]

b、写个程序将对象序列化并输出（ObjectOutputStream能把Object输出成Byte流。将Byte流暂时存储到serial.out文件里）

![8eef49c660e6d2f80213aa46d4179b78.png][]

c、使用ObjectInputStream从持久的文件中读取Bytes重建对象

![fb041461b6c064260f2f703642717656.png][]

# 3. Spark序列化 #

> 大部分Spark程序都具有“内存计算”的天性，所以集群中的所有资源：CPU、网络带宽或者是内存都有可能成为Spark程序的瓶颈。  
> 通常情况下，如果数据完全加载到内存那么网络带宽就会成为瓶颈，但是你仍然需要对程序进行优化，  
> 例如采用序列化的方式保存RDD数据以便减少内存使用。  
> 数据序列化不但能提高网络性能还能减少内存使用。
> 
> 几乎所有的资料都显示kryo 序列化方式优于java自带的序列化方式，而且在spark2.\*版本中都是默认采用kryo 序列化。
> 
> 在spark2.0+版本的官方文档中提到：spark默认提供了两个序列化库：Java自身的序列化和Kryo序列化  
> 官网的解释是：java序列化灵活，但是速度缓慢。Kryo序列化速度更快且更紧凑，但是支持的类型较少。  
> 而且spark现在已经默认RDD在shuffle的时候对简单类型使用了Kryo序列化。

## 3.1 Spark通过两种方式来创建序列化器 ##

**a. [Java][]序列化**

> 在默认情况下，Spark采用Java的ObjectOutputStream序列化一个对象。  
> 该方式适用于所有实现了java.io.Serializable的类。  
> 通过继承java.io.Externalizable，你能进一步控制序列化的性能。  
> Java序列化非常灵活，但是速度较慢，在某些情况下序列化的结果也比较大。

**b. [Kryo][]序列化**

> Spark也能使用Kryo（版本2）序列化对象。  
> Kryo 是 Spark 引入的一个外部的序列化工具, 可以增快 RDD 的运行速度,  
> 因为 Kryo 序列化后的对象更小, 序列化和反序列化的速度非常快  
> Kryo不但速度极快, 而且产生的结果更为紧凑（通常能提高10倍）。  
> Kryo的缺点是不支持所有类型, 为了更好的性能, 需要提前注册程序中所使用的类（class）。

## 3.2 如何使用kryo序列化 ##

> 可以在创建SparkContext之前，通过调用System.setProperty("spark.serializer", "spark.KryoSerializer")，  
> 将序列化方式切换成Kryo。  
> 但是Kryo需要用户进行注册，这也是为什么Kryo不能成为Spark序列化默认方式的唯一原因，  
> 但是建议对于任何“网络密集型”（network-intensive)的应用，  
> 都采用这种方式进行序列化方式。  
> [Kryo][Kryo 1]文档描述了很多便于注册的高级选项，例如添加用户自定义的序列化代码。
> 
> 如果对象非常大，还需要增加属性spark.kryoserializer.buffer.mb的值。  
> 该属性的默认值是32，但是该属性需要足够大以便能够容纳需要序列化的最大对象。
> 
> 如果不注册相应的类，Kryo仍然可以工作，  
> 但是需要为了每一个对象保存其对应的全类名（full class name),这非常浪费。

spark中已经包含了kryo库，使用kryo只需要注册即可。官网只提供了scala版本的，java版本的如下：

![20180525234546191][]

或者：System.setProperty("spark.serializer", "org.apache.spark.serializer.KryoSerializer");

![83b1f2d6db4b4c8e156414a9f302d7c1.png][]

两者都可以用，但是测试好像没起到什么效果。于是需要手动注册：

![2018052523530240][]

上图是关于kryo 的一些配置，可以单独注册自己的一个类，如紫色框线部分；  
也可以像红色框线部分一样，自定义一个接口实现类MyKryoRegistrator，在这个类里面将所需的类全部注册。具体操作如下图：  
如果需要序列化的类太多，就在这里逐一列举即可，当然被注册的类要实现java.io.Serializable，即：class TestKryo implements Serializable

![20180525235511870][]

**序列化的效果:**

test程序验证效果：

![20180526000206139][]

## 3.5 测试结果： ##

> 三种不同情况下的的RDD大小：  
> 默认不序列化：2017.0 KB  
> 在MyKryoRegistrator中序列化: 960.2 KB  
> 只序列化demo.TestKryo:1053.0 KB

# 4. Spark 序列化和反序列化的应用场景 #

### **a.Task 分发** ###

![在这里插入图片描述][20201208152255726.png]

Task 是一个对象, 想在网络中传输对象就必须要先序列化

### **b.RDD 缓存** ###

val rdd1 = rdd.flatMap(_.split(" "))
       .map((_, 1))
       .reduceByKey(_ + _)
    rdd1.cache
    rdd1.collect

>  *  RDD 中处理的是对象, 例如说字符串, Person 对象等
>  *  如果缓存 RDD 中的数据, 就需要缓存这些对象
>  *  对象是不能存在文件中的, 必须要将对象序列化后, 将二进制数据存入文件

### **c.广播变量** ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA3MTE0OTU_size_16_color_FFFFFF_t_70]

广播变量会分发到不同的机器上, 这个过程中需要使用网络, 对象在网络中传输就必须先被序列化

### **d. Shuffle 过程** ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA3MTE0OTU_size_16_color_FFFFFF_t_70 1]

*  Shuffle 过程是由 Reducer 从 Mapper 中拉取数据, 这里面涉及到两个需要序列化对象的原因

>  *  RDD 中的数据对象需要在 Mapper 端落盘缓存, 等待拉取
>  *  Mapper 和 Reducer 要传输数据对象

*  Spark Streaming 的 Receiver

Spark Streaming 中获取数据的组件叫做 Receiver, 获取到的数据也是对象形式, 在获取到以后需要落盘暂存, 就需要对数据对象进行序列化

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA3MTE0OTU_size_16_color_FFFFFF_t_70 2]

### d. 算子引用外部对象 ###

class userserializable(i: Int)
    
    rdd.map(i => new Unserializable(i))
       .collect
       .foreach(println)

>  *  在 Map算子的函数中, 传入了一个 Unserializable 的对象
>  *  Map 算子的函数是会在整个集群中运行的, 那 Unserializable 对象就需要跟随 Map 算子的函数被传输到不同的节点上
>  *  如果 Unserializable 不能被序列化, 则会报错

### e. RDD 的序列化 ###

![在这里插入图片描述][20201208153229661.png]

*  RDD 的序列化

> RDD 的序列化只能使用 Java 序列化器, 或者 Kryo 序列化器

*  为什么?

> RDD 中存放的是数据对象, 要保留所有的数据就必须要对对象的元信息进行保存, 例如对象头之类的 保存一整个对象, 内存占用和效率会比较低一些

*  Kryo序列化示例

val conf = new SparkConf()
        .setMaster("local[2]")
        .setAppName("KyroTest")
      
      conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      conf.registerKryoClasses(Array(classOf[Person]))
      
      val sc = new SparkContext(conf)
      
      rdd.map(arr => Person(arr(0), arr(1), arr(2)))

### f.DataFrame 和 Dataset 中的序列化 ###

**历史的问题**

> RDD 中无法感知数据的组成, 无法感知数据结构, 只能以对象的形式处理数据

**DataFrame 和 Dataset 的特点:**

*  DataFrame 和 Dataset 是为结构化数据优化的
 *  在 DataFrame 和 Dataset 中, 数据和数据的 Schema 是分开存储的

spark.read
         .csv("...")
         .where($"name" =!= "")
         .groupBy($"name")
         .map(row: Row => row)
         .show()

*  DataFrame 中没有数据对象这个概念, 所有的数据都以行的形式存在于 Row 对象中, Row 中记录了每行数据的结构, 包括列名, 类型等

![在这里插入图片描述][20201208153551418.png]

*  Dataset 中上层可以提供有类型的 API, 用以操作数据, 但是在内部, 无论是什么类型的数据对象 Dataset 都使用一个叫做 InternalRow 的类型的对象存储数据

val dataset: Dataset[Person] = spark.read.csv(...).as[Person]

### 总结 ###

> 1.  当需要将对象缓存下来的时候, 或者在网络中传输的时候, 要把对象转成二进制, 在使用的时候再将二进制转为对象, 这个过程叫做序列化和反序列化
> 2.  在 Spark 中有很多场景需要存储对象, 或者在网络中传输对象
>     
>     1.  Task 分发的时候, 需要将任务序列化, 分发到不同的 Executor 中执行
>     2.  缓存 RDD 的时候, 需要保存 RDD 中的数据
>     3.  广播变量的时候, 需要将变量序列化, 在集群中广播
>     4.  RDD 的 Shuffle 过程中 Map 和 Reducer 之间需要交换数据
>     5.  算子中如果引入了外部的变量, 这个外部的变量也需要被序列化
> 3.  RDD 因为不保留数据的元信息, 所以必须要序列化整个对象, 常见的方式是 Java 的序列化器, 和 Kyro 序列化器
> 4.  Dataset 和 DataFrame 中保留数据的元信息, 所以可以不再使用 Java 的序列化器和 Kyro 序列化器, 使用 Spark 特有的序列化协议, 生成 UnsafeInternalRow 用以保存数据, 这样不仅能减少数据量, 也能减少序列化和反序列化的开销, 其速度大概能达到 RDD 的序列化的 20 倍左右

[Spark]: https://www.jianshu.com/p/e1e19aa51eeb
[spark_kryo]: https://blog.csdn.net/sinadrew/article/details/80457854
[Spark 1]: https://blog.csdn.net/u010711495/article/details/110874182
[81899018d926236430971b75fbbf802e.png]: /images/20221023/ecd69875a9e74ebd9e92fc69275da0be.png
[8eef49c660e6d2f80213aa46d4179b78.png]: /images/20221023/3b81ca89d0ca494caf31362086ad3187.png
[fb041461b6c064260f2f703642717656.png]: /images/20221023/7e956fb865614cec87ec26cb1560df6b.png
[Java]: https://link.jianshu.com?t=http://docs.oracle.com/javase/6/docs/api/java/io/Serializable.html
[Kryo]: https://link.jianshu.com?t=http://code.google.com/p/kryo/wiki/V1Documentation
[Kryo 1]: https://link.jianshu.com?t=http://code.google.com/p/kryo/
[20180525234546191]: /images/20221023/b8f07fb586e649568c008f2a619130d3.png
[83b1f2d6db4b4c8e156414a9f302d7c1.png]: /images/20221023/a1dccd80ec2a49578a0f53da0d87ce8b.png
[2018052523530240]: /images/20221023/154a15d8d18b45ccbab8e1d283979209.png
[20180525235511870]: /images/20221023/c9ae9a04bd7449cfb5c3536006b0d102.png
[20180526000206139]: /images/20221023/b51e4dd322354674b7094d846828976c.png
[20201208152255726.png]: /images/20221023/c6ba3f1f9f094415bd87b8062d8b4dbc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA3MTE0OTU_size_16_color_FFFFFF_t_70]: /images/20221023/118b50b183314e478f7306309feb4f56.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA3MTE0OTU_size_16_color_FFFFFF_t_70 1]: /images/20221023/91eee92cc68841d49f2cd1fae3c8ca2e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA3MTE0OTU_size_16_color_FFFFFF_t_70 2]: /images/20221023/7ec449fb8b214b58b4bb7fcaf10c621a.png
[20201208153229661.png]: /images/20221023/cf79b2e92d1c4bc3a6c9fecab6dae9df.png
[20201208153551418.png]: /images/20221023/bcf4928a09ea488da06ad37b0a835897.png