Spark系列--SparkCore(五)RDD容错性之cache、persist、checkpoint

悠悠 2022-05-19 03:21 248阅读 0赞

## 一、RDD容错性 ##

理解RDD的容错性：

RDD可以通过血统机制来进行RDD的恢复。在RDD进行转换和动作的时候，会形成RDD的Lineage依赖链，当某一个RDD失效的时候，可以通过重新计算上游的RDD来重新生成丢失的RDD数据。

但是在spark计算里面,假如计算流程DAG特别长,服务器需要将整个DAG计算完成得出结果,但是如果在这很长的计算流程中突然中间算出的数据丢失了,spark又会根据RDD的依赖关系从头到尾计算一遍,这样子就很费性能,当然我们可以将中间的计算结果通过cache或者persist放到内存或者磁盘中,但是这样也不能保证数据完全不会丢失,存储的这个内存出问题了或者磁盘坏了,也会导致spark从头再根据RDD计算一遍,所以就有了checkpoint,其中checkpoint的作用就是将DAG中比较重要的中间数据做一个检查点将结果存储到一个高可用的地方(通常这个地方就是HDFS里面)。

**后面会详细讲述cache、persist、checkpoint的详细用法。**

--------------------

## 二、cache、persist ##

Spark最重要的一个功能是它可以通过各种操作（operations）持久化（或者缓存）一个集合到内存中。当你持久化一个RDD的时候，每一个节点都将参与计算的所有分区数据存储到内存中，并且这些数据可以被这个集合（以及这个集合衍生的其他集合）的动作（action）重复利用。这个能力使后续的动作速度更快（通常快10倍以上）。对应迭代算法和快速的交互使用来说，缓存是一个关键的工具。

你能通过persist()或者cache()方法持久化一个rdd。首先，在action中计算得到rdd；然后，将其保存在每个节点的内存中。Spark的缓存是一个容错的技术-如果RDD的任何一个分区丢失，它可以通过原有的转换（transformations）操作自动的重复计算并且创建出这个分区。

此外，我们可以利用不同的存储级别存储每一个被持久化的RDD。例如，它允许我们持久化集合到磁盘上、将集合作为序列化的Java对象持久化到内存中、在节点间复制集合或者存储集合到Tachyon(分布式内存文件系统)中。我们可以通过传递一个StorageLevel对象给persist()方法设置这些存储级别。cache()方法使用了默认的存储级别—StorageLevel.MEMORY\_ONLY。完整的存储级别介绍如下所示：

<table> 
 <thead> 
  <tr> 
   <th>级别</th> 
   <th align="center">使用的内存空间</th> 
   <th align="center">CPU时间</th> 
   <th align="center">备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>MEMORY_ONLY</td> 
   <td align="center">高</td> 
   <td align="center">低</td> 
   <td align="center">将RDD作为非序列化的Java对象存储在jvm中。如果RDD不适合存在内存中，一些分区将不会被缓存，从而在每次需要这些分区时都需重新计算它们。这是系统默认的存储级别。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DISK</td> 
   <td align="center">高</td> 
   <td align="center">中等</td> 
   <td align="center">将RDD作为非序列化的Java对象存储在jvm中。如果RDD不适合存在内存中，将这些不适合存在内存中的分区存储在磁盘中，每次需要时读出它们。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY_SER</td> 
   <td align="center">低</td> 
   <td align="center">高</td> 
   <td align="center">将RDD作为序列化的Java对象存储（每个分区一个byte数组）。这种方式比非序列化方式更节省空间，特别是用到快速的序列化工具时，但是会更耗费cpu资源—密集的读操作。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_AND_DISK_SER</td> 
   <td align="center">低</td> 
   <td align="center">高</td> 
   <td align="center">和MEMORY_ONLY_SER类似，但不是在每次需要时重复计算这些不适合存储到内存中的分区，而是将这些分区存储到磁盘中。</td> 
  </tr> 
  <tr> 
   <td>DISK_ONLY</td> 
   <td align="center">低</td> 
   <td align="center">高</td> 
   <td align="center">仅仅将RDD分区存储到磁盘中。</td> 
  </tr> 
  <tr> 
   <td>MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc. (replicated)</td> 
   <td align="center">-</td> 
   <td align="center">-</td> 
   <td align="center">和上面的存储级别类似，但是复制每个分区到集群的两个节点上面。</td> 
  </tr> 
  <tr> 
   <td>OFF_HEAP (experimental)</td> 
   <td align="center">-</td> 
   <td align="center">-</td> 
   <td align="center">以序列化的格式存储RDD到Tachyon中。相对于MEMORY_ONLY_SER，OFF_HEAP减少了垃圾回收的花费，允许更小的执行者共享内存池。这使其在拥有大量内存的环境下或者多并发应用程序的环境中具有更强的吸引力。</td> 
  </tr> 
 </tbody> 
</table>

Spark也会自动持久化一些shuffle操作（如reduceByKey）中的中间数据，即使用户没有调用persist方法。这样的好处是避免了在shuffle出错情况下，需要重复计算整个输入。如果用户计划重用计算过程中产生的RDD，我们仍然推荐用户调用persist方法。

--------------------

**如何选择存储级别**

Spark的多个存储级别意味着在内存利用率和cpu利用效率间的不同权衡。我们推荐通过下面的过程选择一个合适的存储级别：

*  如果你的RDD适合默认的存储级别（MEMORY\_ONLY），就选择默认的存储级别。因为这是cpu利用率最高的选项，会使RDD上的操作尽可能的快。
 *  如果不适合用默认的级别，选择MEMORY\_ONLY\_SER。选择一个更快的序列化库提高对象的空间使用率，但是仍能够相当快的访问。
 *  除非函数计算RDD的花费较大或者它们需要过滤大量的数据，不要将RDD存储到磁盘上，否则，重复计算一个分区就会和重复从磁盘上读取数据一样慢。
 *  如果你希望更快的错误恢复，可以利用重复(replicated)存储级别。所有的存储级别都可以通过血统机制重新计算丢失的数据来支持完整的容错，但是重复的数据能够使你在RDD上继续运行任务，而不需要重新计算丢失的数据。

在拥有大量内存的环境中或者多应用程序的环境中，OFF\_HEAP具有如下优势：

*  它运行多个执行者共享Tachyon中相同的内存池。
 *  它显著地减少垃圾回收的花费。
 *  如果单个的执行者崩溃，缓存的数据不会丢失。

--------------------

**如何使用缓存**

*  调用rdd.persist();变量可以这样设置 如：rdd.persist(StorageLevel.MEMORY\_ONLY); 这里使用了MEMORY\_ONLY级别存储。当然也可以选择其他的如： rdd.persist(StorageLevel.DISK\_ONLY());
 *  调用rdd.cache()方法，cache()是rdd.persist(StorageLevel.MEMORY\_ONLY)的简写，效果和他一模一样的。
 *  调用rdd.unpersist()清除缓存

**注：Spark自动的监控每个节点缓存的使用情况，利用最近最少使用原则删除老旧的数据。如果你想手动的删除RDD，可以使用RDD.unpersist()方法。**

**附：StorageLevel 源码**

val NONE = new StorageLevel(false, false, false, false)
      val DISK_ONLY = new StorageLevel(true, false, false, false)
      val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
      val MEMORY_ONLY = new StorageLevel(false, true, false, true)
      val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
      val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
      val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
      val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
      val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
      val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
      val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
      val OFF_HEAP = new StorageLevel(true, true, true, false, 1)

**说明：上面”\_2”代表的是份数,就是把持久化的数据存为2份**

StorageLevel有五个属性分别是

private var _useDisk: Boolean, //useDisk_是否使用磁盘
        private var _useMemory: Boolean, //useMemory_是否使用内存
        private var _useOffHeap: Boolean, //useOffHeap_是否使用堆外内存如：Tachyon，
        private var _deserialized: Boolean,//deserialized_是否进行反序列化
        private var _replication: Int = 1) //replication_备份数目。

--------------------

## 三、Checkpoint ##

前面提到过，cache或者persist放到内存或者磁盘中,但是这样也不能保证数据完全不会丢失,存储的这个内存出问题了或者磁盘坏了,也会导致spark从头再根据RDD计算一遍,所以就有了checkpoint,其中checkpoint的作用就是将DAG中比较重要的中间数据做一个检查点将结果存储到一个高可用的地方(通常这个地方就是HDFS里面)。

**checkpoint特性**

*  **checkpoint是transformation，当遇到action时，checkpoint会启动另一个任务，将数据切割拆分，保存到设置的checkpoint目录中。**
 *  **当使用了checkpoint后，数据被保存到HDFS，此RDD的依赖关系也会丢掉，因为数据已经持久化到硬盘，不需要重新计算。**
 *  **强烈推荐先将数据持久化到内存中（cache操作），否则直接使用checkpoint会开启一个计算，浪费资源。**

**如何使用checkpoint**

首先需要用sparkContext设置hdfs的checkpoint的目录(如果不设置使用checkpoint会抛出异常:throw new SparkException(“Checkpoint directory has not been set in the SparkContext”):

scala> sc.setCheckpointDir("hdfs://hadoop0:9000/checkpoint")

checkpoint的目录设置好后，如下使用：

rdd.cache()
    rdd.checkpoint()
    rdd.collect