spark RDD详解

Posted 2022-12-11 大数据最好

   RDD(Resilient Distributed Dataset)是Spark的最基本抽象，是对分布式内存的抽象使用，实现了以操作本地集合的方式来操作分布式数据集的抽象实现。RDD是Spark最核心的东西，它表示已被分区，不可变的并能够被并行操作的数据集合，不同的数据集格式对应不同的RDD实现。RDD必须是可序列化的。RDD可以cache到内存中，每次对RDD数据集的操作之后的结果，都可以存放到内存中，下一个操作可以直接从内存中输入，省去了MapReduce大量的磁盘IO操作。这对于迭代运算比较常见的机器学习算法, 交互式数据挖掘来说，效率提升比较大。

RDD特性：

RDD是分布式只读且已分区集合对象。这些集合是弹性的，如果数据集一部分丢失，则可以对它们进行重建。具有自动容错、位置感知调度和可伸缩性，而容错性是最难实现的，大多数分布式数据集的容错性有两种方式：数据检查点和记录数据的更新。对于大规模数据分析系统，数据检查点操作成本很高，主要原因是大规模数据在服务器之间的传输带来的各方面的问题，相比记录数据的更新，RDD 也只支持粗粒度的转换，也就是记录如何从其它 RDD 转换而来（即 Lineage），以便恢复丢失的分区。
其特性为：

1.数据存储结构不可变

2.支持跨集群的分布式数据操作

3.可对数据记录按key进行分区

4.提供了粗粒度的转换操作

5.数据存储在内存中，保证了低延迟性

RDD好处：

• RDD只能从持久存储或通过Transformations操作产生，相比于分布式共享内存(DSM)可以更高效实现容错，对于丢失部分数据分区只需根据它的lineage就可重新计算出来，而不需要做特定的Checkpoint。
• RDD的不变性，可以实现类Hadoop MapReduce的推测式执行。
• RDD的数据分区特性，可以通过数据的本地性来提高性能，这与Hadoop MapReduce是一样的。
• RDD都是可序列化的，在内存不足时可自动降级为磁盘存储，把RDD存储于磁盘上，这时性能会有大的下降但不会差于现在的MapReduce。

RDD编程接口：

对于RDD，有两种类型的动作，一种是Transformation，一种是Action。它们本质区别是：

Transformation返回值还是一个RDD。它使用了链式调用的设计模式，对一个RDD进行计算后，变换成另外一个RDD，然后这个RDD又可以进行另外一次转换。这个过程是分布式的
Action返回值不是一个RDD。它要么是一个Scala的普通集合，要么是一个值，要么是空，最终或返回到Driver程序，或把RDD写入到文件系统中

Transformations转换操作，返回值还是一个 RDD，如 map、 filter、 union；
Actions行动操作，返回结果或把RDD持久化起来，如 count、 collect、 save。

RDD依赖关系：

不同的操作依据其特性，可能会产生不同的依赖，RDD之间的依赖关系有以下两种：

• 窄依赖(Narrow Dependencies)
  一个父RDD分区最多被一个子RDD分区引用，表现为一个父RDD的分区；
  对应于一个子RDD的分区或多个父RDD的分区对应于一个子RDD的分区，也就是说一个父RDD的一个分区不可能对应一个子RDD的多个分区，如map、filter、union等操作则产生窄依赖；
• 宽依赖(Wide Dependencies)
  一个子RDD的分区依赖于父RDD的多个分区或所有分区，也就是说存在一个父RDD的一个分区对应一个子RDD的多个分区，如groupByKey等操作则产生宽依赖操作；

下图中，蓝色实心方框代表一个partition，蓝边矩形框代表一个RDD：

Stage DAG：

Spark提交Job之后会把Job生成多个Stage，多个Stage之间是有依赖的，Stage之间的依赖关系就构成了DAG（有向无环图）。
对于窄依赖，Spark会尽量多地将RDD转换放在同一个Stage中；而对于宽依赖，但大多数时候是shuffle操作，因此Spark会将此Stage定义为ShuffleMapStage，以便于向MapOutputTracker注册shuffle操作。Spark通常将shuffle操作定义为stage的边界。

RDD数据存储管理：

RDD可以被抽象地理解为一个大的数组（Array），但是这个数组是分布在集群上的。逻辑上RDD的每个分区叫一个Partition。
在Spark的执行过程中，RDD经历一个个的Transfomation算子之后，最后通过Action算子进行触发操作。 逻辑上每经历一次变换，就会将RDD转换为一个新的RDD，RDD之间通过Lineage产生依赖关系，这个关系在容错中有很重要的作用。变换的输入和输出都是RDD。 RDD会被划分成很多的分区分布到集群的多个节点中。分区是个逻辑概念，变换前后的新旧分区在物理上可能是同一块内存存储。 这是很重要的优化，以防止函数式数据不变性（immutable）导致的内存需求无限扩张。有些RDD是计算的中间结果，其分区并不一定有相应的内存或磁盘数据与之对应，如果要迭代使用数据，可以调cache()函数缓存数据。

上图中，RDD1含有5个分区（p1、 p2、 p3、 p4、 p5），分别存储在4个节点（Node1、 node2、 Node3、 Node4）中。RDD2含有3个分区（p1、 p2、 p3），分布在3个节点（Node1、 Node2、 Node3）中。

在物理上，RDD对象实质上是一个元数据结构，存储着Block、 Node等的映射关系，以及其他的元数据信息。一个RDD就是一组分区，在物理数据存储上，RDD的每个分区对应的就是一个Block，Block可以存储在内存，当内存不够时可以存储到磁盘上。
每个Block中存储着RDD所有数据项的一个子集，暴露给用户的可以是一个Block的迭代器（例如，用户可以通过mapPartitions获得分区迭代器进行操作），也可以就是一个数据项（例如，通过map函数对每个数据项并行计算）。本书会在后面章节具体介绍数据管理的底层实现细节。
如果是从HDFS等外部存储作为输入数据源，数据按照HDFS中的数据分布策略进行数据分区，HDFS中的一个Block对应Spark的一个分区。同时Spark支持重分区，数据通过Spark默认的或者用户自定义的分区器决定数据块分布在哪些节点。例如，支持Hash分区（按照数据项的Key值取Hash值，Hash值相同的元素放入同一个分区之内）和Range分区（将属于同一数据范围的数据放入同一分区）等分区策略。

• RDD只能从持久存储或通过Transformations操作产生，相比于分布式共享内存(DSM)可以更高效实现容错，对于丢失部分数据分区只需根据它的lineage就可重新计算出来，而不需要做特定的Checkpoint。
• RDD的不变性，可以实现类Hadoop MapReduce的推测式执行。
• RDD的数据分区特性，可以通过数据的本地性来提高性能，这与Hadoop MapReduce是一样的。
• RDD都是可序列化的，在内存不足时可自动降级为磁盘存储，把RDD存储于磁盘上，这时性能会有大的下降但不会差于现在的MapReduce。

RDD编程接口

对于RDD，有两种类型的动作，一种是Transformation，一种是Action。它们本质区别是：

Transformation返回值还是一个RDD。它使用了链式调用的设计模式，对一个RDD进行计算后，变换成另外一个RDD，然后这个RDD又可以进行另外一次转换。这个过程是分布式的
Action返回值不是一个RDD。它要么是一个Scala的普通集合，要么是一个值，要么是空，最终或返回到Driver程序，或把RDD写入到文件系统中

Transformations转换操作，返回值还是一个 RDD，如 map、 filter、 union；
Actions行动操作，返回结果或把RDD持久化起来，如 count、 collect、 save。

RDD依赖关系

不同的操作依据其特性，可能会产生不同的依赖，RDD之间的依赖关系有以下两种：

• 窄依赖(Narrow Dependencies)
  一个父RDD分区最多被一个子RDD分区引用，表现为一个父RDD的分区；
  对应于一个子RDD的分区或多个父RDD的分区对应于一个子RDD的分区，也就是说一个父RDD的一个分区不可能对应一个子RDD的多个分区，如map、filter、union等操作则产生窄依赖；
• 宽依赖(Wide Dependencies)
  一个子RDD的分区依赖于父RDD的多个分区或所有分区，也就是说存在一个父RDD的一个分区对应一个子RDD的多个分区，如groupByKey等操作则产生宽依赖操作；

下图中，蓝色实心方框代表一个partition，蓝边矩形框代表一个RDD：

Stage DAG

Spark提交Job之后会把Job生成多个Stage，多个Stage之间是有依赖的，Stage之间的依赖关系就构成了DAG（有向无环图）。
对于窄依赖，Spark会尽量多地将RDD转换放在同一个Stage中；而对于宽依赖，但大多数时候是shuffle操作，因此Spark会将此Stage定义为ShuffleMapStage，以便于向MapOutputTracker注册shuffle操作。Spark通常将shuffle操作定义为stage的边界。

RDD数据存储管理

RDD可以被抽象地理解为一个大的数组（Array），但是这个数组是分布在集群上的。逻辑上RDD的每个分区叫一个Partition。
在Spark的执行过程中，RDD经历一个个的Transfomation算子之后，最后通过Action算子进行触发操作。 逻辑上每经历一次变换，就会将RDD转换为一个新的RDD，RDD之间通过Lineage产生依赖关系，这个关系在容错中有很重要的作用。变换的输入和输出都是RDD。 RDD会被划分成很多的分区分布到集群的多个节点中。分区是个逻辑概念，变换前后的新旧分区在物理上可能是同一块内存存储。 这是很重要的优化，以防止函数式数据不变性（immutable）导致的内存需求无限扩张。有些RDD是计算的中间结果，其分区并不一定有相应的内存或磁盘数据与之对应，如果要迭代使用数据，可以调cache()函数缓存数据。

上图中，RDD1含有5个分区（p1、 p2、 p3、 p4、 p5），分别存储在4个节点（Node1、 node2、 Node3、 Node4）中。RDD2含有3个分区（p1、 p2、 p3），分布在3个节点（Node1、 Node2、 Node3）中。

在物理上，RDD对象实质上是一个元数据结构，存储着Block、 Node等的映射关系，以及其他的元数据信息。一个RDD就是一组分区，在物理数据存储上，RDD的每个分区对应的就是一个Block，Block可以存储在内存，当内存不够时可以存储到磁盘上。
每个Block中存储着RDD所有数据项的一个子集，暴露给用户的可以是一个Block的迭代器（例如，用户可以通过mapPartitions获得分区迭代器进行操作），也可以就是一个数据项（例如，通过map函数对每个数据项并行计算）。本书会在后面章节具体介绍数据管理的底层实现细节。
如果是从HDFS等外部存储作为输入数据源，数据按照HDFS中的数据分布策略进行数据分区，HDFS中的一个Block对应Spark的一个分区。同时Spark支持重分区，数据通过Spark默认的或者用户自定义的分区器决定数据块分布在哪些节点。例如，支持Hash分区（按照数据项的Key值取Hash值，Hash值相同的元素放入同一个分区之内）和Range分区（将属于同一数据范围的数据放入同一分区）等分区策略。