Spark RDD机制（持久化、依赖关系、checkpoint）

一、RDD持久化

1.1 RDD持久化原理    

       RDD采用惰性求值，每次调用行动算子操作、都会从头开始计算。为了避免重复计算的开销，都可以让spark对数据集进行持久化。当对RDD执行持久化操作时，每个节点都会将自己操作的RDD的partition持久化到内存中，并且在之后对该RDD的反复使用中，直接使用内存缓存的partition，对于迭代式算法和快速交互式应用来说，RDD持久化，是非常重要的。

      要持久化一个RDD，只要调用其cache()或者persist()方法即可。在该RDD第一次被计算出来时，就会直接缓存在每个节点中。而且Spark的持久化机制还是自动容错的，如果持久化的RDD的任何partition丢失了，那么Spark会自动通过其源RDD，使用transformation操作重新计算该partition。
        cache()和persist()的区别在于，cache()是persist()的一种简化方式，cache()的底层就是调用的persist()的无参版本，同时就是调用persist(MEMORY_ONLY)，将数据持久化到内存中。如果需要从内存中清除缓存，那么可以使用unpersist()方法。

1.2 RDD持久化策略

      RDD持久化是可以手动选择不同的策略的。比如可以将RDD持久化在内存中、持久化到磁盘上、使用序列化的方式持久化，多持久化的数据进行多路复用。只要在调用persist()时传入对应的StorageLevel即可。

1.3 如何选择RDD持久化策略？

       Spark提供的多种持久化级别，主要是为了在CPU和内存消耗之间进行取舍。下面是一些通用的持久化级别的选择建议：
       1、优先使用MEMORY_ONLY，如果可以缓存所有数据的话，那么就使用这种策略。因为纯内存速度最快，而且没有序列化，不需要消耗CPU进行反序列化操作。
       2、如果MEMORY_ONLY策略，无法存储的下所有数据的话，那么使用MEMORY_ONLY_SER，将数据进行序列化进行存储，纯内存操作还是非常快，只是要消耗CPU进行反序列化。
       3、如果需要进行快速的失败恢复，那么就选择带后缀为_2的策略，进行数据的备份，这样在失败时，就不需要重新计算了。
       4、能不使用DISK相关的策略，就不用使用，有的时候，从磁盘读取数据，还不如重新计算一次。

1.4 注意事项

• 必须在transformation或者textFile等创建了一个RDD之后，直接连续调用cache()或persist()才可以；        
•  如果你先创建一个RDD，然后单独另起一行执行cache()或persist()方法，是没有用的，而且会报错，大量的文件会丢失。

 二、RDD的依赖关系

不同的RDD之间具有依赖关系，RDD与它所依赖的RDD的依赖关系有两种，分别是宽依赖和窄依赖，如果父RDD分区对应1个子RDD的分区就是窄依赖，否则就是宽依赖。

2.1 窄依赖

窄依赖是指1个父RDD分区对应1个子RDD的分区。换句话说，一个父RDD的分区对应于一个子RDD的分区，或者多个父RDD的分区对应于一个子RDD的分区。所以窄依赖又可以分为两种情况：

• 1个子RDD的分区对应于1个父RDD的分区，比如map，filter，union等算子
• 1个子RDD的分区对应于N个父RDD的分区，比如co-partioned join

2.2 宽依赖

宽依赖是指1个父RDD分区对应多个子RDD分区。宽依赖有分为两种情况

• 1个父RDD对应非全部多个子RDD分区，比如groupByKey，reduceByKey，sortByKey
• 1个父RDD对应所有子RDD分区，比如未经协同划分的join

  

三、checkpoint机制

3.1概念

checkpoint产生就是为了更加可靠的持续化机制.通过spark中额sparkContext对象调用setCheckpoint（）方法，设置一个容错文件系统目录（如hdfs）,将checkpoint数据写入之前设置的容错文件系统中进行高可用的持久化存储，，要是后面有节点出现宕机导致分区数据丢失，可以从作为检查点的rdd开始重新计算，不需要进行从头到尾的计算，减少开销

 3.2  RDD之checkpoint流程

1. SparkContext设置checkpoint目录，用于存放checkpoint的数据； 对RDD调用checkpoint方法，然后它就会被RDDCheckpointData对象进行管理，此时这个RDD的 checkpoint状态会被设置为Initialized
2. 待RDD所在的job运行结束，会调用job中最后一个RDD的doCheckpoint方法，该方法沿着RDD的血 缘 关 系 向 上 查 找 被 checkpoint() 方 法 标 记 过 的 RDD ， 并 将 其 checkpoint 状 态 从 Initialized 设 置 为checkpointingInProgress
3. 启动一个单独的job，来将血缘关系中标记为CheckpointInProgress的RDD执行checkpoint操作，也 就是将其数据写入checkpoint目录
4. 将RDD数据写入checkpoint目录之后，会将RDD状态改变为Checkpointed； 并且还会改变RDD的血缘关系，即会清除掉RDD所有依赖的RDD； 最后还会设置其父RDD为新创建的CheckpointRDD

 3.3 checkpoint与持久化的区别

1. lineage是否发生改变 linage（血缘关系）说的就是RDD之间的依赖关系 持久化，只是将数据保存在内存中或者本地磁盘文件中，RDD的lineage(血缘关系)是不变的； Checkpoint执行之后，RDD就没有依赖的RDD了，也就是它的lineage改变了
2. 丢失数据的可能性 持久化的数据丢失的可能性较大，如果采用 persist 把数据存在内存中的话，虽然速度最快但是也是 最不可靠的，就算放在磁盘上也不是完全可靠的，因为磁盘也会损坏。 Checkpoint的数据通常是保存在高可用文件系统中(HDFS),丢失的可能性很低

建议：对需要checkpoint的RDD，先执行persist(StorageLevel.DISK_ONLY)