Spark RDD弹性分布式数据集——理论

傷城~ 2022-11-17 10:15 238阅读 0赞

### Spark RDD 理论 ###

*   *  1.1 RDD简介
     *  1.2 RDD五大特征
     *  1.3 RDD的处理过程
     *  1.4 RDD的分区
     *   *   *  1.4.1 原因:
             *  1.4.2 目的:
             *  1.4.3 RDD分区的一个分区原则是：
             *  1.4.4 Spark包含两种数据分区方式：
     *  1.5 RDD的依赖关系
     *   *   *  1.5.1 窄依赖
             *  1.5.2 宽依赖
             *  1.5.3 stage的划分
             *  1.5.4 DAG(有向无环图)
     *  1.6 持久化机制
     *  1.7 容错机制
     *   *   *  1.7.1 血统( Lineage )方式
             *  1.7.2 设置检查点(checkpoint)方式
     *  1.8 Spark的任务调度

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

[Hadoop集群搭建及配置⑤ —— Zookeeper 讲解及安装][Hadoop_ _ Zookeeper]

[Hadoop集群搭建及配置⑥ —— Hadoop组件安装及配置][Hadoop_ _ Hadoop]

[Hadoop集群搭建及配置⑦—— Spark&Scala安装配置][Hadoop_ Spark_Scala]

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

[Spark 交互式下实现词频统计][Spark]

[Spark RDD弹性分布式数据集——理论][Spark RDD]

[Spark RDD弹性分布式数据集——实践][Spark RDD 1]

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

Mysql+Hive：[1、Centos7 MySQL安装 —— 用网盘简单安装][1_Centos7 MySQL_ _]

[2、Hadoop集群搭建及配置⑨——Hive 可靠的安装配置][2_Hadoop_Hive]

[3、Spark SQ操作 MySQL数据库和 Hive数据仓库][3_Spark SQ_ MySQL_ Hive]

[4、Spark SQL RDD基本操作、RDD—DataFrame、API MySQL][4_Spark SQL RDD_RDD_DataFrame_API MySQL]

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

传统的MapReduce虽然具有**自动容错、平衡负载和可拓展性**的优点，但是其最大缺点是采用非循环式的数据流模型，使得在迭代计算式要进行大量的磁盘 **I/O** 操作。 Spark中的RDD可以很好的解决这一缺点。

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

## 1.1 RDD简介 ##

*  RDD ( Resilient Distributed Dataset ) ，即**弹性分布式数据集**，是一个**容错的、并行的数据结构**，可以让用户显式地将数据存储到磁盘和内存中，并且还能控制数据的分区。
 *  RDD是Spark提供的最重要的抽象概念，我们可以将RDD理解为一个分布式存储在集群中的大型数据集合，不同RDD之间可以通过转换操作形成依赖关系实现管道化，从而避免了中间结果的 I/O 操作，提高数据处理的速度和性能。
 *  RDD是Spark中的抽象数据结构类型，实现了以操作本地集合的方式来操作分布式数据集的抽象实现，它是一个**不可变、可分区**且里面的元素可**并行计算**的集合。
 *  RDD是**可序列化**的，允许用户在执行多个查询时显式地将工作集缓存在内存中，后续的查询能够重用工作集，这极大地提升了查询速度，省去了MapReduce大量的磁盘IO操作。
 *  任何数据在Spark中都被表示为RDD。从编程的角度来看，RDD可以简单看成是一个数组。和普通数组的区别是，RDD中的数据是分区存储的可以分布在不同的机器上，同时可以被并行处理。
 *  Spark应用程序所做的无非是把需要处理的数据转换为RDD，然后对RDD进行一系列的变换和操作从而得到结果。

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

## 1.2 RDD五大特征 ##

1.  **分区列表**（a list of partitions）  
    每个RDD被分为多个分区(Partitions)，这些分区运行在集群的不同节点上，每个分区都会被一个计算任务处理，分区数决定并行计算的数量。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目(设置的最大core数)。
2.  **每个分区都要一个计算函数**(a function for computing each split)  
    一个计算每个分区的函数。RDD逻辑上是分区的，每个分区的数据是抽象存在的，计算的时候会通过一个compute函数得到每个分区的数据。
    
    Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数，对具体的分区进行计算。compute函数会对迭代器进行复合，不需要保存每次计算的结果。
3.  **依赖于其他RDD**
    
    RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系【窄依赖(一对一)，宽依赖(多对多)】。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。
4.  **(Key,Value) 数据类型的RDD分区器** (a Partitioner for Key-Value RDDs)
    
    当Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于(Key,Value)的RDD，才会有Partitioner，非(Key,Value)的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。
5.  **分区优先位置列表**  
    优先位置列表会存储存取每个Partition的优先位置。对于一个HDFS文件来说，这个列表存储的是每个Partition所在块的位置。按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。

**【注】** RDD是只读的，要想改变RDD中的数据，只能创建一个新的RDD由一个RDD转换到另一个RDD，通过操作算子（map、filter、union、join、reduceByKey… …）实现。

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

## 1.3 RDD的处理过程 ##

Spark用Scala语言实现了RDD的API ，程序开发者可以通过调用API对RDD进行操作处理。RDD经过一系列的“转换”操作，每一次转换都会产生不同的RDD，以供给下一次"转换"操作使用，直到最后一个RDD经过 **"行动"操作** 才会被真正计算处理( 惰性调用 )，并输出到外部数据源中，若是中间的数据结果需要复用，则可以进行缓存处理，将数据缓存到内存中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70]

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

## 1.4 RDD的分区 ##

在分布式程序中，网络通信的开销是很大的，因此控制数据分布以获得最少的网络传输可以极大的提升程序的整体性能， Spark程序可以通过控制RDD分区方式来减少**通信开销**。Spark中所有的RDD都可以进行分区，系统会根据一个针对键的函数对元素进行分区。虽然Spark不能控制每个键具体划分到哪个节点上，但是可以确保相同的键出现在同一个分区上。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 1]  
对于不同的Spark部署模式而言（本地模式、Standalone模式、YARN模式、Mesos模式），都可以通过设置**spark.default.parallelism**这个参数的值，来配置默认的分区数目。

*  本地模式：默认为本地机器的CPU数目，若设置了local\[N\],则默认为N 。
 *  Standalone或YARN：在“集群中所有CPU核心数目总和”和“2”二者中取较大值作为默认值。
 *  Apache Mesos：默认的分区数为8。

#### 1.4.1 原因: ####

要处理的原始数据很大，会被分成很多个分区，分别保存在不同的节点上。

#### 1.4.2 目的: ####

设置合理的并行度，提高数据处理的性能。

#### 1.4.3 RDD分区的一个分区原则是： ####

1.  尽可能使得分区的个数，等于集群核心数目；
2.  尽可能使同一 RDD 不同分区内的记录的数量一致。

#### 1.4.4 Spark包含两种数据分区方式： ####

**HashPartitioner(哈希分区)** 和 **RangePartitioner(范围分区)**，数据分区方式只作用于(Key,Value) 类型数据。

**HashPartitioner(哈希分区)**  
HashPartitioner采用哈希的方式对<Key，Value>键值对数据进行分区。其数据分区规则为 **`partitionId = Key.hashCode % numPartitions`**，其中partitionId代表该Key对应的键值对数据应当分配到的Partition标识，**Key.hashCode** 表示该Key的哈希值，**numPartitions**表示包含的Partition个数。

**RangePartitioner(范围分区)**  
Spark引入RangePartitioner的目的是为了解决HashPartitioner所带来的**分区倾斜**问题，也即分区中包含的数据量不均衡问题。HashPartitioner采用哈希的方式将同一类型的Key分配到同一个Partition中，因此当某一或某几种类型数据量较多时，就会造成若干Partition中包含的数据过大问题，而在Job执行过程中，一个Partition对应一个Task，此时就会使得某几个Task运行过慢。RangePartitioner基于抽样的思想来对数据进行分区。

**【注】:**

1.  只有Key-Value类型的RDD才有分区器，非Key-Value类型的RDD分区器的值是None。
2.  如果是对键操作，则子RDD不再继承父RDD的分区器，但是分区数会继承。

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

## 1.5 RDD的依赖关系 ##

**作用：** 1. 用来解决数据容错；2. 用来划分stage。

#### 1.5.1 窄依赖 ####

*  窄依赖是指父RDD的每一个分区最多被一个子RDD的分区使用，即OneToOneDependencies。
 *  窄依赖的表现一般分为两类，第一类表现为一个父RDD的分区对应于一个子RDD的分区;第二类表现为多个父RDD的分区。对应于一个子RDD的分区。**(1:1 n:1)**
 *  一个父RDD的一个分区不可能对应一个子RDD的多个分区。
 *  为了便于理解，我们通常把窄依赖形象的比喻为独生子女。

RDD做map，filter和union算子操作时，是属于窄依赖的第一类表现；而RDD做join算子操作(对输入进行协同划分)时，是属于窄依赖表现的第二类。输入协同划分是指多个父RDD的某一个分区的所有Key ，被划分到子RDD的同一分区。当子RDD做算子操作，因为某个分区操作失败导致数据丢失时，只需要重新对父RDD中对应的分区做算子操作即可恢复数据。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 2]

#### 1.5.2 宽依赖 ####

宽依赖是指子RDD的每一个分区都会使用所有父RDD的所有分区或多个分区，即OneToManyDependecies，为了便于理解，我们通常把宽依赖形象的比喻为超生。**(1:n n:n)**

父RDD做groupByKey和join (输入未协同划分)算子操作时，子RDD的每一个分区都会依赖于所有父RDD的所有分区。当子RDD做算子操作，因为某个分区操作失败导致数据丢失时，则需要重新对父RDD中的所有分区进行算子操作才能恢复数据。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 3]

#### 1.5.3 stage的划分 ####

action触发job，依照RDD依赖关系切分若干TaskSet即(stage) ; task任务处理的最小单元。

**划分Stage：**

1.  从后向前遍历，遇到宽依赖就断开，遇到窄依赖就把当前的RDD加入到Stage中；
2.  每个Stage里面的Task的数量是由该Stage中最后一个RDD的Partition数量决定的；
3.  最后一个Stage里面的任务类型是ResultTask，前面所有其他Stage里面的任务类型都是ShuffleMapTask；
4.  代表当前Stage的算子一定是该Stage的最后一个计算步骤。

**用宽依赖划分stage原因：**

1.  相比于宽依赖，窄依赖对优化很有利；
2.  宽依赖对应着shuffle操作，数据传输时宽依赖对应多个节点的传输；
3.  当RDD分区丢失时（某个节点故障），spark会对数据进行重算时,宽依赖重算的效用不仅在于算的多，还在于有多少是冗余的计算；
4.  窄依赖能够更有效地进行失效节点的恢复，即只需重新计算丢失RDD分区的父分区，而且不同节点之间可以并行计算；
5.  窄依赖允许在一个集群节点上以流水线的方式（pipeline）计算所有父分区。

#### 1.5.4 DAG(有向无环图) ####

DAG ( Directed Acyclic Graph )叫做有向无环图，Spark中的RDD通过一系列的转换算子操作和行动算子操作形成了一个DAG。DAG是一种非常重要的图论数据结构。如果一个有向图无法从任意顶点出发经过若干条边回到该点，则这个图就是有向无环图。"4→6 → 1→2”是一条路径，"4 →6 →5"也是一条路径，并且图中不存在从顶点经过若干条边后能回到该点。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 4]  
根据RDD之间依赖关系的不同可将DAG划分成不同的Stage(调度阶段)。对窄依赖来说，RDD分区的转换处理是在一个线程里完成，所以窄依赖会被Spark划分到同一个Stage中；

对宽依赖来说，由于有Shuffle存在，所以只能在父RDD处理完成后，下一个Stage才能开始接下来的计算，因此宽依赖是划分Stage的依据，当RDD进行转换操作，遇到宽依赖类型的转换操作时，就划为一个Stage。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 5]  
**A、 C、E 是上图三个RDD的实例：**

*  当A做groupByKey转换操作生成B时，由于groupByKey转换操作属于宽依赖类型，所以就把A划分为一个Stage，如Stage1 ；
 *  当C做map转换操作生成D， D与E做union转换操作生成F。由于map和union转换操作都属于窄依赖类型，因此不进行Stage的划分，而是将C、D、E。F加入到同一个Stage中；
 *  当F与B进行join转换操作时，由于这时的join操作是非协同划分，所以属于宽依赖，因此会划分为一个Stage ，如Stage2 ；
 *  剩下的B和G被划分为一个Stage ，如Stage3。

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

## 1.6 持久化机制 ##

*  如果使用默认的存储级别(MEMORY ONLY) ，存储在内存中的RDD没有发生溢出，那么就选择默认的存储级别。默认存储级别可以最大程度的提高CPU的效率，可以使在RDD上的操作以最快的速度运行。
 *  如果内存不能全部存储RDD，那么使用MEMORY-ONLY-SER，并挑选一个快速序列化库将对象序列化，以节省内存空间。使用这种存储级别，计算速度仍然很快。
 *  除了在计算该数据集的代价特别高，或者在需要过滤大量数据的情况下，尽量不要将溢出的数据存储到磁盘。因为，重新计算这个数据分区的耗时与从磁盘读取这些数据的耗时差不多。
 *  如果想快速还原故障，建议使用多副本存储级别(例如，使用Spark作为web应用的后台服务，在服务出故障时需要快速恢复的场景下) 。所有的存储级别都通过重新计算丢失的数据的方式，提供了完全容错机制。但是多副本级别在发生数据丢失时，不需要重新计算对应的数据库，可以让任务继续运行。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 6]  
    在Spark中，RDD是采用**惰性求值**，即每次调用行动算子操作，都会从头开始计算，这对迭代计算来说代价很大，因为迭代计算经常需要多次重复的使用同一组数据集，所以，为了避免重复计算的开销，让Spark对数据集进行持久化操作。

RDD的持久化操作有两种方法，分别是 **cache()** 方法和 **persist()** 方法。

*  persist() 方法的存储级别是通过StorageLevel对象设置的。用persist()方法对一个RDD标记为持久化，出现persist()语句的地方，并不会马上计算生成RDD并把它持久化，而是要等到遇到第一个行动操作触发真正计算以后，才会把计算结果进行持久化。持久化后的RDD将会被保留在计算节点的内存中被后面的行动操作重复使用。

**persist()的括号包含的是持久化级别参数：**

*  **persist(MEMORY\_ONLY)**：表示将RDD作为反序列化的对象存储于JVM中，如果内存不足，就要按照LRU原则替换缓存中的内容。
 *  **persist(MEMORY\_AND\_DISK)**：表示将RDD作为反序列化的对象存储在JVM中，如果内存不足，超出的分区将会被存放在硬盘上。
 *  可以使用unpersist()方法手动地把持久化的RDD从缓存中  
    移除
 *  cache() 方法的存储级别是使用默认的存储级别(即StorageLevel.MEMORY\_ONLY)。

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

## 1.7 容错机制 ##

当Spark集群中的某一个节点由于宕机导致数据丢失，则可以通过Spark中的RDD进行容错恢复已经丢失的数据。RDD提供了两种**故障恢复**的方式，分别是**血统( Lineage )方式**和**设置检查点(checkpoint)方式**。

#### 1.7.1 血统( Lineage )方式 ####

根据RDD之间依赖关系对丢失数据的RDD进行数据恢复。若丢失数据的子RDD进行窄依赖运算，则只需要把丢失数据的父RDD的对应分区进行重新计算，不依赖其他节点，并且在计算过程中不存在冗余计算；若丢失数据的RDD进行宽依赖运算，则需要父RDD所有分区都要进行从头到尾计算，计算过程中存在冗余计算。

#### 1.7.2 设置检查点(checkpoint)方式 ####

本质是将RDD写入磁盘存储。当RDD进行宽依赖运算时，只要在中间阶段设置一个检查点进行容错，即Spark中的sparkContext调用**setCheckpoint()** 方法，设置容错文件系统目录作为检查点checkpoint ，将checkpoint的数据写入之前设置的容错文件系统中进行持久化存储，若后面有节点宕机导致分区数据丢失，则以从做检查点的RDD开始重新计算，不需要从头到尾的计算，从而减少开销。

## 1.8 Spark的任务调度 ##

Spark的任务调度流程：即RDD在Spark中的运行流程分为RDD Objects、DAGScheduler、TaskScheduler以及Worker四个部分。

1.  **RDD Objects**：当RDD对象创建后， SparkContext会根据RDD对象构建DAG有向无环图，然后将Task提交给DAGScheduler。
2.  **DAGScheduler**：将作业的DAG划分成不同Stage ，每个Stage都是TaskSet任务集合，并以TaskSet为单位提交给TaskScheduler。
3.  **TaskScheduler**：通过TaskSetManager管理Task，并通过集群中的资源管理器把Task发给集群中Worker的Executor。
4.  **Worker:** Spark集群中的Worker接收到Task后，把Task运行在Executor进程中，一个进程中可以有多个线程在工作，从而可以处理多个数据分区。

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

————————————————

\[1\] 参考博文：[https://blog.csdn.net/qq\_45710900/article/details/102473976][https_blog.csdn.net_qq_45710900_article_details_102473976]

[Hadoop_ _ Zookeeper]: https://blog.csdn.net/weixin_44775255/article/details/113772666
[Hadoop_ _ Hadoop]: https://blog.csdn.net/weixin_44775255/article/details/113775449
[Hadoop_ Spark_Scala]: https://blog.csdn.net/weixin_44775255/article/details/114518207
[Spark]: https://blog.csdn.net/weixin_44775255/article/details/114694265
[Spark RDD]: https://blog.csdn.net/weixin_44775255/article/details/115536606
[Spark RDD 1]: https://blog.csdn.net/weixin_44775255/article/details/115560667
[1_Centos7 MySQL_ _]: https://blog.csdn.net/weixin_44775255/article/details/115662187
[2_Hadoop_Hive]: https://blog.csdn.net/weixin_44775255/article/details/115663334
[3_Spark SQ_ MySQL_ Hive]: https://blog.csdn.net/weixin_44775255/article/details/115707963
[4_Spark SQL RDD_RDD_DataFrame_API MySQL]: https://blog.csdn.net/weixin_44775255/article/details/115727069
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70]: /images/20221022/b9780b2b5515455ba9e959c9736a5904.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 1]: /images/20221022/a770b6d1dd204a8b8fb1ac2d3b57df9e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 2]: /images/20221022/536f39e2e0d74960b5d93890d9804a72.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 3]: /images/20221022/19ebbd34ba8243a49b3c6090625b19c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 4]: /images/20221022/f4b3d16d4cb4451cab34c610fbdced69.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 5]: /images/20221022/e9b6aa6acbef416f9da008eeab433e4f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 6]: /images/20221022/6aa6852eb0fb4d11a78c1cece80064ad.png
[https_blog.csdn.net_qq_45710900_article_details_102473976]: https://blog.csdn.net/qq_45710900/article/details/102473976