什么是持久化？

持久化的意思就是说将RDD的数据缓存到内存中或者持久化到磁盘上，只需要缓存一次，后面对这个RDD做任何计算或者操作，可以直接从缓存中或者磁盘上获得，可以大大加快后续RDD的计算速度。

为什么要持久化？

在之前的文章中讲到Spark中有tranformation和action两类算子，tranformation算子具有lazy特性，只有action算子才会触发job的开始，从而去执行action算子之前定义的tranformation算子，从hdfs中读取数据等，计算完成之后，Spark会将内存中的数据清除，这样处理的好处是避免了OOM问题，但不好之处在于每次job都会从头执行一边，比如从hdfs上读取文件等，如果文件数据量很大，这个过程就会很耗性能。这个问题就涉及到本文要讲的RDD持久化特性，合理的使用RDD持久化对Spark的性能会有很大提升。

即：默认情况下，action算子执行后，会会将内存中的数据（RDD）清除，防止OOM

RDD持久化原理

Spark可以将RDD持久化在内存中。当对RDD执行持久化操作时，每个节点都会将自己操作的RDD的partition数据持久化到内存中，并且在之后对该RDD的反复使用中，直接使用内存中缓存的partition数据。

这样的话，针对一个RDD反复执行多个操作的场景，就只需要对RDD计算一次即可，后面直接使用该RDD，而不需要反复计算多次该RDD。
因为正常情况下这个RDD的数据使用过后内存中是不会一直保存的。

例如这样的操作：针对mapRDD需要多次使用的，没有开启持久化时，每次action算子执行后，mapRDD、dataRDD都会被清除，防止OOM

val dataRDD = sc.parallelize(Array(1,2,3,4,5))     
val mapRDD = dataRDD.map(...)     
mapRDD.foreach(...)     
mapRDD.saveAsTextFile(...)     
mapRDD.collect()

巧妙使用RDD持久化，在某些场景下，对spark应用程序的性能有很大提升。特别是对于迭代式算法和快速交互式应用来说，RDD持久化，是非常重要的。

cache()和persist()

要持久化一个RDD，只需要调用它的cache()或者persist()方法就可以了。
在该RDD第一次被计算出来时，就会直接缓存在每个节点中。而且Spark的持久化机制还是自动容错的，如果持久化的RDD的任何partition数据丢失了，那么Spark会自动通过其源RDD，使用transformation算子重新计算该partition的数据。

cache()和persist()的区别：
cache()是persist()的一种简化方式，cache()的底层就是调用的persist()的无参版本，也就是调用persist(MEMORY_ONLY)，将数据持久化到内存中。

unpersist()

如果需要从内存中清除缓存，那么可以使用unpersist()方法。

RDD持久化策略

下面看一下目前Spark支持的一些持久化策略

策略	介绍
MEMORY_ONLY	以非序列化的方式持久化在JVM内存中
MEMORY_AND_DISK	同上，但是当某些partition无法存储在内存中时，会持久化到磁盘中
MEMORY_ONLY_SER	同MEMORY_ONLY，但是会序列化
MEMORY_AND_DISK_SER	同MEMORY_AND_DSK，但是会序列化
DISK_ONLY	以非序列化的方式完全存储到磁盘上
MEMORY_ONLY_2、MEMORY_AND_DISK_2等	如果是尾部加了2的持久化级别，表示会将持久化数据复用一份，保存到其他节点

序列化：把Java对象转换为字节序列的过程。
反序列化：把字节序列恢复为Java对象的过程。

MEMORY_ONLY（★）

以非序列化的方式持久化在JVM内存中

如果内存无法完全存储RDD所有的partition，那么那些没有持久化的partition就会在下一次需要使用它的时候，重新被计算。

MEMORY_AND_DISK

同上，但是当某些partition无法存储在内存中时，会持久化到磁盘中

下次需要使用这些partition时，需要从磁盘上读取，不需要重新计算

MEMORY_ONLY_SER

同MEMORY_ONLY，但是会序列化

将Java对象序列化后进行持久化。可以减少内存开销，但是在使用的时候需要进行反序列化，因此会增加CPU开销。

MEMORY_AND_DISK_SER

同MEMORY_AND_DSK，但是会序列化

DISK_ONLY

以非序列化的方式完全存储到磁盘上

MEMORY_ONLY_2、MEMORY_AND_DISK_2等

如果是尾部加了2的持久化级别，表示会将持久化数据复制一份，保存到其它节点，从而在数据丢失时，不需要重新计算，只需要使用备份数据即可。

如何选择RDD持久化策略

Spark提供了多种持久化级别，主要是为了在CPU和内存消耗之间进行取舍。
下面是一些通用的持久化级别的选择建议：

优先使用MEMORY_ONLY，纯内存速度最快，而且没有序列化不需要消耗CPU进行反序列化操作，缺点就是比较耗内存
MEMORY_ONLY_SER，将数据进行序列化存储，纯内存操作还是非常快，只是在使用的时候需要消耗CPU进行反序列化

注意：如果需要进行数据的快速失败恢复，那么就选择带后缀为_2的策略，进行数据的备份，这样在失败时，就不需要重新计算了能不使用DISK相关的策略，就不要使用，因为有的时候，从磁盘读取数据，还不如重新计算一次。

案例：使用RDD的持久化

object PersistRddScala {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("PersistRddScala")
    conf.setMaster("local")
    val sc = new SparkContext(conf)

    //这里通过textFile()方法，针对外部文件创建了一个RDD，实际上，程序执行到这里为止，hello_10000000.dat文件的数据是不会被加载到内存中的。需要等到action算子（count()）执行才开始
    //cache默认是基于内存的持久化，调用的是persist()，persist(StorageLevel.MEMORY_ONLY)，注意cache的用法和位置
    val dataRDD = sc.textFile("D:\\test\\hello_10000000.dat").cache()
    var start_time = System.currentTimeMillis()
    var count = dataRDD.count()
    println(count)
    var end_time = System.currentTimeMillis()
    println("第一次耗时：" + (end_time - start_time))


    start_time = System.currentTimeMillis()
    count = dataRDD.count()
    println(count)
    end_time = System.currentTimeMillis()
    println("第二次耗时：" + (end_time - start_time))

    sc.stop()
  }

}

没有开启cache()，每一次都耗时很长，因为textFile()需要加载2次（2次的原因是：action算子执行后，Spark会将内存中的数据清除，防止OOM）

开启cache()，第二次计算耗时就很少了，textFile()加载1次

共享变量

共享变量的工作原理

Spark还有一个非常重要的特性就是共享变量
默认情况下，如果在一个算子函数中使用到了某个外部的变量，那么这个变量的值会被拷贝到每个task中。此时每个task只能操作自己的那份变量数据。如果多个task想要共享某个变量，那么这种方式是做不到的。
Spark为此提供了两种共享变量
一种是Broadcast Variable（广播变量）
另一种是Accumulator（累加变量）

Broadcast Variable（广播变量）

Broadcast Variable会将使用到的变量，仅仅为每个节点拷贝一份，而不会为每个task都拷贝一份副本，因此其最大的作用，就是减少变量到各个节点的网络传输消耗，以及在各个节点上的内存消耗。
通过调用SparkContext的broadcast()方法，针对某个变量创建广播变量
注意：广播变量，是只读的
然后在算子函数内，使用到广播变量时，每个节点只会拷贝一份副本。可以使用广播变量的value()方法获取值。
接下来看一个图深入理解一下

先看左边的代码
这个是一个咱们经常使用的map算子的代码，map算子中执行对每一个元素乘以一个固定变量的操作，此时这个固定的变量属于外部变量。
默认情况下，算子函数内，使用到的外部变量，会被拷贝到执行这个算子的每一个task中
看图中间的MapTask，这些都是map算子产生的task，也就是说这个外部变量会被拷贝到每一个task中。
如果这个外部变量是一个集合，集合中有上亿条数据，这个网络传输就会很耗时，而且在每个task上，占用的内存空间，也会很大

如果算子函数中使用的外部变量，是广播变量的话，那么每个变量只会拷贝一份到每个节点上。节点上所有的task都会共享这一份变量，就可以减少网络传输消耗的时间，以及减少内存占用了。
大家可以想象一个极端情况，如果map算子有10个task，恰好这10个task还都在一个worker节点上，那么这个时候，map算子使用的外部变量就会在这个worker节点上保存10份，这样就很占用内存了。

下面我们来具体使用一下这个广播变量
Scala代码如下：

object BroadCastOpScala {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("BroadCastOpScala")
    conf.setMaster("local")
    val sc = new SparkContext(conf)

    val rdd = sc.parallelize(Array(1, 2, 3, 4, 5))
    val varable = 2
    //rdd.map(_ * varable).foreach(println(_))

    //1、定义广播变量
    val varableBroadCast = sc.broadcast(varable)

    //2、使用广播变量，调用其value方法
    rdd.map(_ * varableBroadCast.value).foreach(println(_))

    sc.stop()
  }
}

Accumulator（累加变量）

Spark提供的Accumulator，主要用于多个节点对一个变量进行共享性的操作。

正常情况下在Spark的任务中，由于一个算子可能会产生多个task并行执行，所以在这个算子内部执行的聚合计算都是局部的，想要实现多个task进行全局聚合计算，此时需要使用到Accumulator这个共享的累加变量。
注意：Accumulator只提供了累加的功能。在task只能对Accumulator进行累加操作，不能读取它的值。只有在Driver进程中才可以读取Accumulator的值。

下面我们来写一个案例

object AccumulatorOpScala {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("AccumulatorOpScala")
      .setMaster("local")
    val sc = new SparkContext(conf)

    val dataRDD = sc.parallelize(Array(1, 2, 3, 4, 5))

    /**
     * 这种写法是错误的，因为foreach代码是在worker节点上执行的
     * var total = 0 和 println(total) 是在Driver进程中执行的，所以无法实现累加操作
     * 并且foreach算子可能会在多个task中执行，这样foreach内部实现的累加也不是最终全局累加的结果
     */
    /*var total = 0
    dataRDD.foreach(num=>total += num)
    println(total)*/

    //所以此时想要实现累加操作就需要使用累加变量了
    //1：定义累加变量
    val sumAccumulator = sc.longAccumulator

    //2：使用累加变量
    dataRDD.foreach(num => sumAccumulator.add(num))

    //注意：只能在Driver进程中获取累加变量的结果
    println(sumAccumulator.value)

    sc.stop()
  }

}

运行结果