rdd缓存

Ⅰ Spark核心-RDD

RDD是Spark中的数据抽象，全称 弹性分布式数据集（Resilient Distributed Datasets） 。RDD可以理解为将一个大的数据集合以分布式的形式保存在集群服务器的内存中。RDD是一个容错的、并行的数据结构，可以让用户显式地将数据存储到磁盘和内存中，并能控制数据的分区。

RDD是Spark的核心，也是整个Spark的架构基础。

RDD的特点：

RDD的5个主要属性：

可以通过两种方式创建RDD：

转换操作指的是在原RDD实例上进行计算，然后创建一个新的RDD实例。

RDD中的所有的转换操作都是 惰性 的，在执行RDD的转换操作的时候，并不会直接计算结果，而是记住这些应用到基础数据集上的转换动作，只有行动操作时，这些转换才会真正的去执行。这样设计的好处是更加有效率的运行。

行动操作指的是向驱动器程序返回结果或把结果写入外部系统的操作。

Spark在调用RDD的行动操作的时候，会触发Spark中的连锁反应。当调用的行动操作的时候，Spark会尝试创建作为调用者的RDD。如果这个RDD是从文件中创建的，那么Spark会在worker节点上读取文件至内存中。如果这个RDD是通过其他RDD的转换得到的，Spark会尝试创建其父RDD。这个过程会一直持续下去，直到Spark找到根RDD。然后Spark就会真正执行这些生成RDD所必须的转换计算。最后完成行动操作，将结果返回给驱动程序或者写入外部存储。

Spark速度非常快的原因之一，就是在不同操作中在内存中持久化一个数据集。当持久化一个RDD后，每一个节点都将把计算的分片结果保存在内存中，并在对此数据集进行的其他动作中重用。这使得后续的动作变得更加迅速。缓存是Spark构建迭代算法和快速交互式查询的关键。所以我们在开发过程中，对经常使用的RDD要进行缓存操作，以提升程序运行效率。

RDD缓存的方法

RDD类提供了两种缓存方法：

cache方法其实是将RDD存储在集群中Worker的内存中。

persist是一个通用的cache方法。它可以将RDD存储在内存中或硬盘上或者二者皆有。

缓存的容错

缓存是有可能丢失（如机器宕机），或者存储于内存的数据由于内存不足而被删除。RDD的缓存的容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于RDD的一系列的转换，丢失的数据会被重新计算。因为RDD的各个Partition是相对独立的，所以在重新计算的时候只需要计算丢失部分Partition即可，不需要重新计算全部的Partition。因此，在一个缓存RDD的节点出现故障的时候，Spark会在另外的节点上自动重新创建出现故障的节点中存储的分区。

RDD的缓存能够在第一次计算完成后，将计算结果保存到内存、本地文件系统或者Tachyon中。通过缓存，Spark避免了RDD上的重复计算，能够极大地提升计算速度。但是，如果缓存丢失了，则需要重新计算。如果计算特别复杂或者计算特别耗时，那么缓存丢失对于整个Job的影响是不容忽视的。为了避免缓存丢失重新计算带来的开销，所以Spark引入了检查点（checkpoint）机制。

缓存是在计算结束后，直接将计算结果通过用户定义的存储级别写入不同的介质。而检查点不同，它是在计算完成后，重新建立一个Job来计算。所以为了避免重复计算，推荐先将RDD缓存，这样在进行检查点操作时就可以快速完成。

Spark会根据用户提交的计算逻辑中的RDD的转换和动作来生动RDD之间的依赖关系，同时这个计算链也就生成了逻辑上的DAG。

RDD之间的依赖关系包括：

Spark中的依赖关系主要体现为两种形式：

Ⅱ Spark内存管理详解（下）——内存管理

弹性分布式数据集（RDD）作为Spark最根本的数据抽象，是只读的分区记录（Partition）的集合，只能基于在稳定物理存储中的数据集上创建，或者在其他已有的RDD上执行转换（Transformation）操作产生一个新的RDD。转换后的RDD与原始的RDD之间产生的依赖关系，构成了血统（Lineage）。凭借血统，Spark保证了每一个RDD都可以被重新恢复。但RDD的所有转换都是惰性的，即只有当一个返回结果给Driver的行动（Action）发生时，Spark才会创建任务读取RDD，然后真正触发转换的执行。

Task在启动之初读取一个分区时，会先判断这个分区是否已经被持久化，如果没有则需要检查Checkpoint或按照血统重新计算。所以如果一个RDD上要执行多次行动，可以在第一次行动中使用persist或cache方法，在内存或磁盘中持久化或缓存这个RDD，从而在后面的行动时提升计算速度。事实上，cache方法是使用默认的MEMORY_ONLY的存储级别将RDD持久化到内存，故缓存是一种特殊的持久化。 堆内和堆外存储内存的设计，便可以对缓存RDD时使用的内存做统一的规划和管理 （存储内存的其他应用场景，如缓存broadcast数据，暂时不在本文的讨论范围之内）。

RDD的持久化由Spark的Storage模块 [1] 负责，实现了RDD与物理存储的解耦合。Storage模块负责管理Spark在计算过程中产生的数据，将那些在内存或磁盘、在本地或远程存取数据的功能封装了起来。在具体实现时Driver端和Executor端的Storage模块构成了主从式的架构，即Driver端的BlockManager为Master，Executor端的BlockManager为Slave。Storage模块在逻辑上以Block为基本存储单位，RDD的每个Partition经过处理后唯一对应一个Block（BlockId的格式为 rdd_RDD-ID_PARTITION-ID ）。Master负责整个Spark应用程序的Block的元数据信息的管理和维护，而Slave需要将Block的更新等状态上报到Master，同时接收Master的命令，例如新增或删除一个RDD。

在对RDD持久化时，Spark规定了MEMORY_ONLY、MEMORY_AND_DISK等7种不同的 存储级别 ，而存储级别是以下5个变量的组合 [2] ：

通过对数据结构的分析，可以看出存储级别从三个维度定义了RDD的Partition（同时也就是Block）的存储方式：

RDD在缓存到存储内存之前，Partition中的数据一般以迭代器（ Iterator ）的数据结构来访问，这是Scala语言中一种遍历数据集合的方法。通过Iterator可以获取分区中每一条序列化或者非序列化的数据项(Record)，这些Record的对象实例在逻辑上占用了JVM堆内内存的other部分的空间，同一Partition的不同Record的空间并不连续。

RDD在缓存到存储内存之后，Partition被转换成Block，Record在堆内或堆外存储内存中占用一块连续的空间。 将Partition由不连续的存储空间转换为连续存储空间的过程，Spark称之为“展开”（Unroll） 。Block有序列化和非序列化两种存储格式，具体以哪种方式取决于该RDD的存储级别。非序列化的Block以一种DeserializedMemoryEntry的数据结构定义，用一个数组存储所有的Java对象，序列化的Block则以SerializedMemoryEntry的数据结构定义，用字节缓冲区（ByteBuffer）来存储二进制数据。每个Executor的Storage模块用一个链式Map结构（LinkedHashMap）来管理堆内和堆外存储内存中所有的Block对象的实例 [6] ，对这个LinkedHashMap新增和删除间接记录了内存的申请和释放。

因为不能保证存储空间可以一次容纳Iterator中的所有数据，当前的计算任务在Unroll时要向MemoryManager申请足够的Unroll空间来临时占位，空间不足则Unroll失败，空间足够时可以继续进行。对于序列化的Partition，其所需的Unroll空间可以直接累加计算，一次申请。而非序列化的Partition则要在遍历Record的过程中依次申请，即每读取一条Record，采样估算其所需的Unroll空间并进行申请，空间不足时可以中断，释放已占用的Unroll空间。如果最终Unroll成功，当前Partition所占用的Unroll空间被转换为正常的缓存RDD的存储空间，如下图2所示。

在 《Spark内存管理详解（上）——内存分配》 的图3和图5中可以看到，在静态内存管理时，Spark在存储内存中专门划分了一块Unroll空间，其大小是固定的，统一内存管理时则没有对Unroll空间进行特别区分，当存储空间不足是会根据动态占用机制进行处理。

由于同一个Executor的所有的计算任务共享有限的存储内存空间，当有新的Block需要缓存但是剩余空间不足且无法动态占用时，就要对LinkedHashMap中的旧Block进行淘汰（Eviction)，而被淘汰的Block如果其存储级别中同时包含存储到磁盘的要求，则要对其进行落盘（Drop），否则直接删除该Block。
存储内存的淘汰规则为：

落盘的流程则比较简单，如果其存储级别符合 _useDisk 为true的条件，再根据其 _deserialized 判断是否是非序列化的形式，若是则对其进行序列化，最后将数据存储到磁盘，在Storage模块中更新其信息。

Executor内运行的任务同样共享执行内存，Spark用一个HashMap结构保存了任务到内存耗费的映射。每个任务可占用的执行内存大小的范围为 1/2N ~ 1/N ，其中N为当前Executor内正在运行的任务的个数。每个任务在启动之时，要向MemoryManager请求申请最少为1/2N的执行内存，如果不能被满足要求则该任务被阻塞，直到有其他任务释放了足够的执行内存，该任务才可以被唤醒。

执行内存主要用来存储任务在执行Shuffle时占用的内存，Shuffle是按照一定规则对RDD数据重新分区的过程，我们来看Shuffle的Write和Read两阶段对执行内存的使用：

在ExternalSorter和Aggregator中，Spark会使用一种叫AppendOnlyMap的哈希表在堆内执行内存中存储数据，但在Shuffle过程中所有数据并不能都保存到该哈希表中，当这个哈希表占用的内存会进行周期性地采样估算，当其大到一定程度，无法再从MemoryManager申请到新的执行内存时，Spark就会将其全部内容存储到磁盘文件中，这个过程被称为溢存(Spill)，溢存到磁盘的文件最后会被归并(Merge)。

Shuffle Write阶段中用到的Tungsten是Databricks公司提出的对Spark优化内存和CPU使用的计划 [4] ，解决了一些JVM在性能上的限制和弊端。Spark会根据Shuffle的情况来自动选择是否采用Tungsten排序。Tungsten采用的页式内存管理机制建立在MemoryManager之上，即Tungsten对执行内存的使用进行了一步的抽象，这样在Shuffle过程中无需关心数据具体存储在堆内还是堆外。每个内存页用一个MemoryBlock来定义，并用 Object obj 和 long offset 这两个变量统一标识一个内存页在系统内存中的地址。堆内的MemoryBlock是以long型数组的形式分配的内存，其 obj 的值为是这个数组的对象引用， offset 是long型数组的在JVM中的初始偏移地址，两者配合使用可以定位这个数组在堆内的绝对地址；堆外的MemoryBlock是直接申请到的内存块，其 obj 为null， offset 是这个内存块在系统内存中的64位绝对地址。Spark用MemoryBlock巧妙地将堆内和堆外内存页统一抽象封装，并用页表(pageTable)管理每个Task申请到的内存页。

Tungsten页式管理下的所有内存用64位的逻辑地址表示，由页号和页内偏移量组成：

有了统一的寻址方式，Spark可以用64位逻辑地址的指针定位到堆内或堆外的内存，整个Shuffle Write排序的过程只需要对指针进行排序，并且无需反序列化，整个过程非常高效，对于内存访问效率和CPU使用效率带来了明显的提升 [5] 。

Spark的存储内存和执行内存有着截然不同的管理方式：对于存储内存来说，Spark用一个LinkedHashMap来集中管理所有的Block，Block由需要缓存的RDD的Partition转化而成；而对于执行内存，Spark用AppendOnlyMap来存储Shuffle过程中的数据，在Tungsten排序中甚至抽象成为页式内存管理，开辟了全新的JVM内存管理机制。

Spark的内存管理是一套复杂的机制，且Spark的版本更新比较快，笔者水平有限，难免有叙述不清、错误的地方，若读者有好的建议和更深的理解，还望不吝赐教。