> For the complete documentation index, see [llms.txt](https://2327257266.gitbook.io/learn-spa/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://2327257266.gitbook.io/learn-spa/spark-rdds/rdds-de-te-dian.md). # RDDs 的特点 由于RDD是一个只读的分区数据集，无法直接对数据集进行修改，只能通过RDD自带的转化操作，由一个RDD得到一个新的RDD，而这个新的RDD就是其他RDD的衍生物，也就肯定包含了其他RDD的信息。 基于这样一个特性，RDD之间必定存在依赖，这使得RDDs会形成一个有**向无环图DAG，**而这个DAG就像是记录这一系列转化过程的记录者，在实际执行时，就可以通过这样的依赖关系进行计算。 > **注意：RDD还可以将数据集缓存到内存中，使得在多个操作之间可以重用数据集，基于这个特点可以很方便地构建迭代型应用(图计算、机器学习等)或者交互式数据分析应用。** 基于这样的介绍，RDD存在以下4个特点： **1、分区性** 对于每一个RDD，其逻辑上是一个分区的，且是该分区上数据的一个抽象形式，只有在计算的时候通过compute函数得到每个分区的数据，这一点有点像深度学习语言，先构建好计算图，在训练时在传入数据。 ![](/files/-MFB2-bA079rl1vsHkiT) 也就是说，如果该RDD是通过已存在的文件进行构建的，则compute函数是读取指定文件系统中的数据；如果RDD是通过其他RDD转换来的，则compute函数是执行转换逻辑将其他RDD的数据进行转换。总的来说，**RDD是分区间的，之间产生的依赖关系通过compute函数去操作。** **2、只读性** 由于RDD是只读的，因此想要改变RDD中的数据集，只能通过RDD的转换创建新的RDD。Spark提供RDD丰富的操作算子，实现一个RDD转化成另一个RDD，如下图所示。 ![](/files/-MFB3_ySnR-TPl4Qe6eH) **3、依赖性** RDDs通过操作算子进行转换，转换得到的新RDD包含了从其他RDDs衍生所必需的信息，RDDs之间维护着这种血缘关系，也称之为依赖。 ![](/files/-MFB3qIkMlNR-vOwoyAg) 如图所示，依赖包括两种，一种是*窄依赖*，RDDs之间分区是一一对应的，另一种是*宽依赖*，下游RDD的每个分区与上游RDD(也称之为父RDD)的每个分区都有关，是多对多的关系。 **其中对于窄依赖在实际执行过程中会类似于管道一样一气呵成，就是说，多不操作之间不进行存储，在执行完之后再进行存储。** **4、缓存** 这是Spark中实现加速的一个手段，即对于一个RDD如果被多个RDD所使用的时候(多个RDD的父RDD)，可以将这个RDD缓存起来，，该RDD只有在第一次计算的时候根据依赖关系得到分区的数据，在后续其他地方用到时，直接从缓存处取，不需要再重新根据依赖关系进行计算，这样就加速后期的重用。 ![](/files/-MFB5j0Gna_wjNp5UbX4) **CheckPoint** > 虽然RDD的血缘关系天然地可以实现容错，当RDD的某个分区数据失败或丢失，可以通过血缘关系重建。但是对于长时间迭代型应用来说，随着迭代的进行，RDDs之间的血缘关系会越来越长，一旦在后续迭代过程中出错，则需要通过非常长的血缘关系去重建，势必影响性能。 > > 为此，RDD支持checkpoint将数据保存到持久化的存储中，这样就可以切断之前的血缘关系，因为checkpoint后的RDD不需要知道它的父RDDs了，它可以从checkpoint处拿到数据。 ![](/files/-MFB8W2CL9jrOjz6QGeN) ### RDD的持久化 由于RDD采用惰性机制，每次遇到行动操作都会根据DAG的依赖关系从头开始执行计算，如果遇到迭代计算，需要重复调用中间数据，会造成极大的计算开销； 可以通过持久化操作来解决以上的问题，用`persist()`方法对需要重复使用的RDD标记为持久化，当遇到第一次行动操作后，会把计算结果持久化，保存在计算节点的内存备用； ### persist() `persist(storageLevel)`：storageLevel参数表示持久化级别，通过使用不同的级别可以把数据缓存到不同的位置，其中使用`cache()`函数会调用默认的持久化方法，即`persist(MEMORY_ONLY)`将RDD作为反序列化的对象存储在JVM中；而`unpersist()`方法则可以把持久化的RDD从缓存中删除。 ``` rdd = sc.parallelize(['spark','rdd','hadoop']) rdd.cache() print(rdd.take(1)) # 第一次行动操作，触发完整计算，同时把rdd放入缓存 # 输出 ['spark'] print(rdd.collect()) # 第二次行动操作，此时直接重复利用上述的缓存rdd # 输出 ['spark','rdd','hadoop'] rdd.unpersist() ``` 除了 `MEMORY_ONLY` 还有不同的级别可以把数据缓存到不同的位置，下表给出所有的存储等级： | Storage Level | Meaning | | ------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | | MEMORY\_ONLY | 将RDD作为非序列化的Java对象存储在jvm中。如果RDD不适合存在内存中，一些分区将不会被缓存，从而在每次需要这些分区时都需重新计算它们。这是系统默认的存储级别。 | | MEMORY\_AND\_DISK | 将RDD作为非序列化的Java对象存储在jvm中。如果RDD不适合存在内存中，将这些不适合存在内存中的分区存储在磁盘中，每次需要时读出它们。 | | MEMORY\_ONLY\_SER | 将RDD作为序列化的Java对象存储（每个分区一个byte数组）。这种方式比非序列化方式更节省空间，特别是用到快速的序列化工具时，但是会更耗费cpu资源—密集的读操作。 | | MEMORY\_AND\_DISK\_SER | 和MEMORY\_ONLY\_SER类似，但不是在每次需要时重复计算这些不适合存储到内存中的分区，而是将这些分区存储到磁盘中。 | | DISK\_ONLY | 仅仅将RDD分区存储到磁盘中 | | MEMORY\_ONLY\_2, MEMORY\_AND\_DISK\_2, etc. | 和上面的存储级别类似，但是复制每个分区到集群的两个节点上面 | | OFF\_HEAP (experimental) | 以序列化的格式存储RDD到[Tachyon](http://tachyon-project.org/)中。相对于MEMORY\_ONLY\_SER，OFF\_HEAP减少了垃圾回收的花费，允许更小的执行者共享内存池。这使其在拥有大量内存的环境下或者多并发应用程序的环境中具有更强的吸引力。 | ### 如何选择存储级别？ 官网上也指出了不同的存储等级如何进行选择的问题，主要有以下几点： 1、如果可以尽量使用默认的存储级别(MEMORY\_ONLY)，因为使用默认存储，cpu利用率最高，使RDD上的操作尽可能的快。 2、如果不适合用默认的级别，选择MEMORY\_ONLY\_SER。选择一个更快的序列化库提高对象的空间使用率，但是仍能够相当快的访问。 3、除非某个RDD的计算需要花费大量的时间，否则不要将RDD存储到磁盘上。 4、如果你希望更快的错误恢复，可以利用重复(replicated)存储级别。 5、在拥有大量内存的环境中或者多应用程序的环境中，使用OFF\_HEAP会更好，因为它运行多个执行者共享Tachyon中相同的内存池，如果单个的执行者崩溃，缓存的数据不会丢失，并且还能显著地较少垃圾回收。