大数据Spark框架 6：RDD的基本介绍

小灰灰 2024-02-20 08:56 46阅读 0赞

### 本章内容： ###

*  1- On Yarn 流程介绍 （理解）
 *  2- RDD的基本介绍 （了解）
 *  3- RDD的构建方式 （掌握）
 *  4- RDD的相关的API(算子) （掌握）
 *  5- 综合案例 （操作）

0. PySpark和Spark的交互流程

*  PySpark 和 Spark程序交互流程：client on Spark集群

1- 首先会在提交的的节点上启动一个Driver程序  
2-Driver启动后， 执行mian函数， 首先创建SparkContext对象（底层是基于PY4J，识别python的构建方式， 将其映射为Java代码）  
  
3- 连接Spark集群的主节点，根据资源配置要求， 向主节点申请资源， 用于启动executor （提交的任务的时候， 可以自定义资源信息）  
  
4- Master接收到资源申请后， 根据申请要求， 进行分配资源， 底层的资源分配方案为FIFO，然后将分配好的资源交给Driver程序 executor1：node1 2个CPU 2GB内存  
executor2： node3 2个CPU 2GB内存  
  
5- Driver连接对应的节点， 通知worker启动executor， 并占用相应资源，worker启动完成后， 反向注册回给Driver程序（已经启动完成）  
  
6- Driver开始处理代码：  
6.1 首先会加载所有的RDD相关的API(算子)， 基于算子之间的依赖关系，形成DAG执行流程图，划分stage阶段，并且确定每个阶段应该运行多少个线程以及每个线程应该交给哪个executor来运行（任务分配）  
6.2 Driver程序通知对应的executor程序， 来执行具体的任务  
6.3 Executor接收到任务信息后， 启动线程， 开始执行处理即可：executor在执行的时候， 由于RDD代码中有大量的Python的函数,Executor是一个JVM程序 ，无法解析Python函数， 此时会调用Python解析器，执行函数， 并将函数结果返回给Executor  
6.4 Executor在运行过程中， 如果发现最终的结果需要返回给Driver，直接返回Driver， 如果不需要返回， 直接输出 结束即可  
6.5 Driver程序监听这个executor执行的状态信息， 当Executor都执行完成后， Driver认为任务运行完成了  
  
7- 当任务执行完成后， Driver执行 sc.stop() 通知Master执行完成， Master回收资源， Driver程序退出即可

PySpark 和 Spark程序交互流程：cluster on Spark集群

1- 首先会将任务提交给Spark集群的主节点（Master）  
2- Master接收到任务信息后， 根据Driver的资源配置信息要求， 选择一个worker节点(有资源的，如果都有随机)来启动Driver程序， 并且占用相对应资源  
3- Driver启动后， 执行main函数， 首先创建SparkContext对象（底层是基于PY4J，识别python的构建方式， 将其映射为Java代码） 创建成功后， 会向Master进行注册， 告知已经启动成功了  
  
4- 根据executor的资源配置要求， 向主节点申请资源， 用于启动executor （提交的任务的时候， 可以自定义资源信息）  
  
5- Master接收到资源申请后， 根据申请要求， 进行分配资源， 底层的资源分配方案为FIFO，然后将分配好的资源交给Driver程序 executor1：node1 2个CPU 2GB内存  
executor2： node3 2个CPU 2GB内存  
  
6- Driver连接对应的节点， 通知worker启动executor， 并占用相应资源，worker启动完成后， 反向注册回给Driver程序（已经启动完成）  
  
7- Driver开始处理代码：  
7.1 首先会加载所有的RDD相关的API(算子)， 基于算子之间的依赖关系，形成DAG执行流程图，划分stage阶段，并且确定每个阶段应该运行多少个线程以及每个线程应该交给哪个executor来运行（任务分配）  
7.2 Driver程序通知对应的executor程序， 来执行具体的任务  
7.3 Executor接收到任务信息后， 启动线程， 开始执行处理即可：executor在执行的时候， 由于RDD代码中有大量的Python的函数,Executor是一个JVM程序 ，无法解析Python函数， 此时会调用Python解析器，执行函数， 并将函数结果返回给Executor  
7.4 Executor在运行过程中， 如果发现最终的结果需要返回给Driver，直接返回Driver， 如果不需要返回， 直接输出 结束即可  
7.5 Driver程序监听这个executor执行的状态信息， 当Executor都执行完成后， Driver认为任务运行完成了  
  
7- 当任务执行完成后， Driver执行 sc.stop() 通知Master执行完成， Master回收资源， Driver程序退出即可

PySpark 和 Spark程序交互流程：client on Yarn集群

区别点：  
将Driver程序中， 负责资源申请的部分， 移交给Yarn中AppMaster来负责 Driver主要负责任务的分配 任务的管理工作  
  
1- 首先会在提交的的节点上启动一个Driver程序  
2-Driver启动后， 执行mian函数， 首先创建SparkContext对象（底层是基于PY4J，识别python的构建方式， 将其映射为Java代码）  
  
3- 连接Yarn集群的主节点,将需要申请的资源包装为一个任务， 提交给Yarn的主节点， 主节点收到任务后， 首先随机的选择一个nodemanager启动AppMaster  
  
4- 当AppMaster启动后， 会和 yarn的主节点建立心跳机制， 告知已经启动成功了， 启动成功后， 就进行资源的申请工作， 将需要申请的资源通过心跳包的形式发送给resourceManager， resourceManager收到资源申请后， 开始进行资源分配工作， 底层是基于资源调度器来实现（默认为: 容量调度）， 当RM将资源分配OK后， 等待AppMaster进行拉取操作， AppMaster会定时的通过心跳的方式询问RM是否已经准备好资源， 一旦发现准备好， 立即拉取对应资源信息  
  
5- AppMaster根据获取到的资源信息连接对应的节点， 通知对应nodemanager启动executor， 并占用相应资源，nodemanage对应executor启动完成后， 反向注册回给Driver程序（已经启动完成）  
  
6- Driver开始处理代码：  
6.1 首先会加载所有的RDD相关的API(算子)， 基于算子之间的依赖关系，形成DAG执行流程图，划分stage阶段，并且确定每个阶段应该运行多少个线程以及每个线程应该交给哪个executor来运行（任务分配）  
6.2 Driver程序通知对应的executor程序， 来执行具体的任务  
6.3 Executor接收到任务信息后， 启动线程， 开始执行处理即可：executor在执行的时候， 由于RDD代码中有大量的Python的函数,Executor是一个JVM程序 ，无法解析Python函数， 此时会调用Python解析器，执行函数， 并将函数结果返回给Executor  
6.4 Executor在运行过程中， 如果发现最终的结果需要返回给Driver，直接返回Driver， 如果不需要返回， 直接输出 结束即可  
6.5 Driver程序监听这个executor执行的状态信息， 当Executor都执行完成后， Driver认为任务运行完成了， 同时AppMaster也会接收到各个节点执行完成状态， 然后通知RM。任务执行完成，RM回收资源， 关闭AppMaster，并通知Driver程序  
  
7- Driver执行 sc.stop() ， Driver程序退出即可

![图片][285ac9aab71e85d9a9870f001f715d4f.png]

PySpark 和 Spark程序交互流程：cluster on Yarn集群

区别点：  
在集群模式下， Driver程序 和 AppMaster 二者合二为一了， Driver就是AppMaster，AppMaster就是Driver  
  
1- 首先会将任务提交给Yarn集群的主节点（resourceManager）  
2- resourceManager接收到任务信息后， 根据Driver(AppMaster)的资源配置信息要求， 选择一个nodeManager节点(有资源的，如果都有随机)来启动Driver(AppMaster)程序， 并且占用相对应资源  
3- Driver(AppMaster)启动后， 执行main函数， 首先创建SparkContext对象（底层是基于PY4J，识别python的构建方式， 将其映射为Java代码） 创建成功后， 会向resourceManager进行建立心跳机制， 告知已经启动成功了  
  
4- 根据executor的资源配置要求， 向主节点通过心跳的方式申请资源， 用于启动executor （提交的任务的时候， 可以自定义资源信息）  
  
5- resourceManager接收到资源申请后， 根据申请要求， 进行分配资源， 底层是基于资源调度器来资源分配（默认容量调度），然后将分配好的资源准备好， 等待Driver(AppMaster)拉取操作  
executor1： node1 2个CPU 2GB内存  
executor2： node3 2个CPU 2GB内存  
  
6- Driver(AppMaster)会定时询问是否准备好资源， 一旦准备好， 立即获取， 根据资源信息连接对应的节点， 通知nodeManager启动executor， 并占用相应资源，nodeManager对应的executor启动完成后， 反向注册回给Driver(AppMaster)程序（已经启动完成）  
  
7- Driver(AppMaster)开始处理代码：  
7.1 首先会加载所有的RDD相关的API(算子)， 基于算子之间的依赖关系，形成DAG执行流程图，划分stage阶段，并且确定每个阶段应该运行多少个线程以及每个线程应该交给哪个executor来运行（任务分配）  
7.2 Driver(AppMaster)程序通知对应的executor程序， 来执行具体的任务  
7.3 Executor接收到任务信息后， 启动线程， 开始执行处理即可：executor在执行的时候， 由于RDD代码中有大量的Python的函数,Executor是一个JVM程序 ，无法解析Python函数， 此时会调用Python解析器，执行函数， 并将函数结果返回给Executor  
7.4 Executor在运行过程中， 如果发现最终的结果需要返回给Driver(AppMaster)，直接返回Driver(AppMaster)， 如果不需要返回， 直接输出 结束即可  
7.5 Driver(AppMaster)程序监听这个executor执行的状态信息， 当Executor都执行完成后，Driver(AppMaster)认为任务运行完成了  
  
7- 当任务执行完成后， Driver执行 sc.stop() 通知RM执行完成， RM回收资源， Driver程序退出即可

![图片][f723a743fe5b3fc3ddc35a3b44ca879e.png]

####  ####

1. RDD的基本介绍

####  ####

1.1 什么是RDD

RDD: 弹性分布式数据集

出现目的：为了能够支持更加高效的迭代计算的操作

背景说明：

早期的计算模型： 单机模型  
例如：Mysql / Excel  
单机的计算模型  
仅适用于： 小量数据集的处理操作  
在计算操作的时候， 只有一个进程， 在一个进程中通过不断的迭代完成最终的计算操作  
  
随着不断的发展， 整个是社会数据量都在不断的增大， 原有的单机的计算模型无法应对未来的数据处理需要， 怎么办？分布式的计算模型  
  
核心： 采用多个节点来处理， 将一个任务拆分为多个子任务， 分别运行在不同的节点上进行并行的计算操作， 各个节点计算完成后， 将结果汇总处理即可 （分而治之）  
诞生了相关的分布式计算框架： MapReduce Spark Flink Storm 。。。。  
  
  
MR 和 Spark都是一个大规模的分布式计算引擎， 都可以处理大规模数据：  
MR存在的弊端：  
1- 执行效率比较低：底层大量的经过磁盘到内存， 内存到磁盘 ， 磁盘到磁盘 。。。IO比较大  
2- MR程序只有Map 和 Reduce， 意味着如何要进行迭代计算操作， 一个MR只能迭代一次 如果想要迭代多次， 那么就必须通过多个MR进行串联， 每一个MR都要申请资源 都要启动和关闭操作， 这些过程相当于都浪费了， 同时整个IO频次更高  
  
正因为MR存在的一些弊端， 对于市场来说， 迫切需要一款能解决MR痛点的架构：  
1- 解决多次磁盘的IO问题， 希望尽可能都在内存中进行处理，减少IO  
2- 提供一个更加高效的迭代计算引擎  
  
RDD的出现就是为了解决这个问题的

MR的迭代模型：

![图片][5f7ad2c71d1ea3be960fdf53e261173f.png]

Spark的迭代计算模型：

![图片][b65e8371cd997a39c8be4ed194d8c2c4.png]

RDD是一个抽象的数据模型， RDD本身并不存储任何的数据， 仅仅是一个数据传输的管道， 在这个管道中， 作为使用者， 只需要告知给RDD应该从哪里读取数据， 中间需要进行什么样的转换逻辑操作，最后需要将结果输出到什么位置即可， 一旦启动后， RDD会根据用户设定的规则， 完成整个处理操作

####  ####

1.2 RDD的五大特性（记忆）

1-（必须的）RDD可分区的  
2-（必须的）每一个RDD都是有一个计算函数  
3-（必须的）RDD之间是存在着依赖关系  
4-（可选的）对于KV类型的时候是存在分区函数的，对于KV类型的RDD默认是基于Hash 分区方案  
5-（可选的）移动数据不如移动计算（让计算程序离数据越近越好）

####  ####

1.3 RDD的五大特点（记忆）

1- RDD是可分区的：分区是一种逻辑分区， 仅仅是定义分区的规则， 并不是直接对数据进行分区操作， 因为RDD本身不存储数据  
2- RDD是只读的：每一个RDD都是不可变的， 如果想要改变， 处理后会得到一个新的RDD，原有RDD保持原样  
3- RDD之间存在依赖关系：每一个RDD之间都是依赖关系， 也称为血缘关系， 一般分为两种依赖关系：宽依赖 和 窄依赖  
4- RDD可以设置cache（缓存）：当计算过程中一个RDD被多个RDD所依赖的时候， 可以将这个RDD的结果缓存起来， 这样后续使用这个RDD的时候， 可以直接获取数据， 不需要在重新计算  
5- checkpoint（检查点）：与缓存类似的， 都是可以将中间某一个RDD的结果保存起来， 只不过checkpoint支持持久化保存

####  ####

2. 如何构建RDD

构建RDD对象的方式主要有二种：

1- 通过 parallelized（data）：通过自定义列表的方式初始化RDD对象 （测试）  
2- 通过 textFile（data）：通过读取外部文件的方式来初始化RDD对象（测试/开发）

![图片][a604780d53fe6e61e0ca31c80f473e6f.png]

####  ####

2.1 通过并行化本地的方式构建RDD

代码实现：

相关的API:

查看分区数量的API:  
rdd.getNumPartitions()  
  
查看每个分区的数据API:  
rdd.glom().collect()  
  
构建RDD的API:  
parallelize(参数1,参数2)：  
参数1： 设置本地列表  
参数2： 表示有多少个分区

通过本地列表的方式构建RDD， 其对应分区数量如何确定的：

1- 默认与 setMaster设置的线程的数量保持一致：比如 local【3】 表示默认有三个分区， 但是如果集群运行， 默认为2  
  
2- local【\*】：\* 对应的数值是与当前节点的CPU核数保持一致  
  
3- 如果想要改变RDD的分区数量：  
方式一： 直接修改 setMaster中local的 N 即可， 但是仅限于 本地模式 如果集群运行 无法修改  
方式二： 修改 parallelize的参数2的位置， 设置多少， 那么就有多少个分区， 一般建议小于等于 setMaster设置

可能出现错误：

java.net.ConnectException: Call From node1.itcast.cn/192.168.88.161 to node1:8020 failed on connection exception: java.net.ConnectException: 拒绝连接; For more details see: http://wiki.apache.org/hadoop/ConnectionRefused  
  
此错误， 原因是无法连接到hadoop， 但是具体的缘由：  
1- hadoop没有启动（大概率）  
2- hadoop没有启动良好（小概率） 这是一般启动不检查最容易出现问题  
3- hadoop本身存在一些问题， 导致无法启动了

####  ####

2.2 通过读取外部数据源方式

TextFile API的方式实现：

说明：分区数量确定， 以看到的现象为准

1- 与 setMaster关系不大， 但是当设置为1的时候， 有用， 往大没用  
2- 可以在textFile中设置最小分区数量， 但是实际这个RDD的分区数量可能会大于等于设置最小分区数量  
3- 默认情况下， 读取多个文件的时候， 有多少个文件， 一般会至少产生多少个分区

处理小文件的操作：

from pyspark import SparkContext, SparkConf  
import os  
  
os.environ\['SPARK\_HOME'\] = '/export/server/spark'  
os.environ\['PYSPARK\_PYTHON'\] = '/root/anaconda3/bin/python3'  
os.environ\['PYSPARK\_DRIVER\_PYTHON'\] = '/root/anaconda3/bin/python3'  
  
\# 程序的入门: pyspark的程序, 必须要有程序入门  
\# 快捷键: main + 回车  
if \_\_name\_\_ == '\_\_main\_\_':  
print("演示如何构建RDD： 方式二 小文件处理方案")  
  
\# 1 创建SparkContext对象  
conf = SparkConf().setAppName('create\_rdd\_01').setMaster('local\[\*\]')  
sc = SparkContext(conf=conf)  
  
\# 2- 读取外部文件数据：小文件问题  
\# rdd = sc.textFile(name='file:///export/data/workspace/gz11\_pyspark\_parent/\_02\_spark\_core/data')  
\# wholeTextFiles 专门用于处理小文件的API， 默认情况下， 尽可能减少分区的数量  
rdd = sc.wholeTextFiles(path='file:///export/data/workspace/gz11\_pyspark\_parent/\_02\_spark\_core/data')  
print(rdd.getNumPartitions())  
print(rdd.glom().collect())  
"""  
\[  
\[  
('file:/export/data/workspace/gz11\_pyspark\_parent/\_02\_spark\_core/data/words.txt', 'hadoop hadoop hadoop hive hive\\r\\nhive hadoop hive sqoop sqoop\\r\\nhadoop hue hue sqoop hive\\r\\nhive hive sqoop hue zookeeper\\r\\nzookeeper hive sqoop oozie\\r\\noozie oozie hive hive'),  
('file:/export/data/workspace/gz11\_pyspark\_parent/\_02\_spark\_core/data/words1.txt', 'hadoop hadoop hadoop hive hive\\r\\nhive hadoop hive sqoop sqoop\\r\\nhadoop hue hue sqoop hive\\r\\nhive hive sqoop hue zookeeper\\r\\nzookeeper hive sqoop oozie\\r\\noozie oozie hive hive'),  
('file:/export/data/workspace/gz11\_pyspark\_parent/\_02\_spark\_core/data/words2.txt', 'hadoop hadoop hadoop hive hive\\r\\nhive hadoop hive sqoop sqoop\\r\\nhadoop hue hue sqoop hive\\r\\nhive hive sqoop hue zookeeper\\r\\nzookeeper hive sqoop oozie\\r\\noozie oozie hive hive')  
\],  
\[  
('file:/export/data/workspace/gz11\_pyspark\_parent/\_02\_spark\_core/data/words3.txt', 'hadoop hadoop hadoop hive hive\\r\\nhive hadoop hive sqoop sqoop\\r\\nhadoop hue hue sqoop hive\\r\\nhive hive sqoop hue zookeeper\\r\\nzookeeper hive sqoop oozie\\r\\noozie oozie hive hive'),  
('file:/export/data/workspace/gz11\_pyspark\_parent/\_02\_spark\_core/data/words4.txt', 'hadoop hadoop hadoop hive hive\\r\\nhive hadoop hive sqoop sqoop\\r\\nhadoop hue hue sqoop hive\\r\\nhive hive sqoop hue zookeeper\\r\\nzookeeper hive sqoop oozie\\r\\noozie oozie hive hive')  
\]  
\]  
  
"""

扩展： 分区数量是如何确定的：

1- 分区数量（线程数量）一般设置为CPU核数2~3倍  
  
2- RDD的分区数量， 取决于多个因素：调用任务的时候， 设置CPU核数， 调用对应API 设置分区数量， 以及本身读取文件分区数  
  
2.1 当初始化SparkContext的时候， 其实我们确定了一个基本的并行度参数：  
参数: spark.default.parallelism  
值: 默认为CPU核数， 如果是集群 至少为2 如果local模式 取决于 local\[N\] N为多少， 即为多少并行度  
  
2.2 如果调用者通过parallelism API 来构建RDD：  
分区数量：  
如果没有指定分区数， 分区数量取决于 spark.default.parallelism  
如果设置分区数量， 取决于自己设置的分区数  
  
2.3 textFile(path,minPartition): 分区的选择  
取决于以下几个变量：  
defaultMinPartition：  
值：  
如果没有指定minPartition，此值为 min(spark.default.parallelism, 2)  
如果指定minPartition，此值等于指定的值  
  
对于读取本地文件来说： 判断标准  
rdd分区数 = max(本地文件分片数，defaultMinPartition)  
对于读取HDFS文件数： 判断标准  
rdd分区数 = max(文件的block块数量，defaultMinPartition)

[285ac9aab71e85d9a9870f001f715d4f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/d8b53583c61b44bea12e7c3ebefe66bf.png
[f723a743fe5b3fc3ddc35a3b44ca879e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/afda0af5d7b04d0fb3a5fa0fac588e63.png
[5f7ad2c71d1ea3be960fdf53e261173f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/57ce79738eb94897a44dcb49c0328293.png
[b65e8371cd997a39c8be4ed194d8c2c4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/600e7991b0a14da297ff2190e41c4415.png
[a604780d53fe6e61e0ca31c80f473e6f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/b6c8526e79374b6e8c43346424f124b6.png