SparkCore（二）：RDD基础

青旅半醒 2022-12-08 04:17 195阅读 0赞

### 文章目录 ###

*  一、什么是RDD
 *  二、RDD核心属性
 *  三、RDD执行原理
 *  四、RDD创建

`Spark`计算框架为了能够进行高并发和高吞吐的数据处理，封装了三大数据结构，用于处理不同的应用场景。三大数据结构分别是：

*  **RDD** : 弹性分布式数据集
 *  **累加器**：分布式共享只写变量
 *  **广播变量**：分布式共享只读变量

# 一、什么是RDD #

`RDD（Resilient Distributed Dataset）`叫做弹性分布式**数据集**，是`Spark`中最基本的数据处理模型。代码中是一个**抽象类**，它代表一个**弹性**的、**不可变**、**可分区**、里面的元素可**并行计算**的集合。

**弹性**：

*  存储的弹性：内存与磁盘的自动切换；
 *  容错的弹性：数据丢失可以自动恢复；
 *  计算的弹性：计算出错重试机制；
 *  分片的弹性：可根据需要重新分片。

**分布式**：数据存储在大数据集群不同节点上；

**数据集**：`RDD`封装了计算逻辑，并不保存数据；

**数据抽象**：`RDD`是一个抽象类，需要子类具体实现；

**不可变**：`RDD`封装了计算逻辑，是不可以改变的，想要改变，只能产生新的`RDD`，在新的`RDD`里面封装计算逻辑；

**可分区、并行计算**；

# 二、RDD核心属性 #

**分区列表**：`RDD`数据结构中存在分区列表，用于执行任务时并行计算，是实现分布式计算的重要属性

**分区计算函数**：`Spark`在计算时，是使用分区函数对每一个分区进行计算

**RDD之间的依赖关系**：`RDD`是计算模型的封装，当需求中需要将多个计算模型进行组合时，就需要将多个`RDD`建立依赖关系

**分区器（可选）**：当数据为`KV`类型数据时，可以通过设定分区器自定义数据的分区

**首选位置（可选）**：计算数据时，可以根据计算节点的状态选择不同的节点位置进行计算

# 三、RDD执行原理 #

从计算的角度来讲，数据处理过程中需要计算资源（内存 & `CPU`）和计算模型（逻辑）。执行时，需要将计算资源和计算模型进行协调和整合。

`Spark`框架在执行时，先申请资源，然后将应用程序的数据处理逻辑分解成一个一个的计算任务。然后将任务发到已经分配资源的计算节点上, 按照指定的计算模型进行数据计算。最后得到计算结果。

`RDD`是`Spark`框架中用于数据处理的核心模型，接下来我们看看，在`Yarn`环境中，`RDD`的工作原理:

1.  启动`Yarn`集群环境  
    ![在这里插入图片描述][20200924102200408.png]
2.  `Spark`通过申请资源创建调度节点和计算节点  
    ![在这里插入图片描述][20200924102249678.png]
3.  `Spark`框架根据需求将计算逻辑根据分区划分成不同的任务  
    ![在这里插入图片描述][20200924102313904.png]
4.  调度节点将任务根据计算节点状态发送到对应的计算节点进行计算  
    ![在这里插入图片描述][20200924102404469.png]

# 四、RDD创建 #

在`Spark`中创建`RDD`的创建方式可以分为四种：

**① 从集合（内存）中创建RDD**

从集合中创建`RDD`，`Spark`主要提供了两个方法：`parallelize()`和`makeRDD()`。

val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDDTest1")
        val sc = new SparkContext(sparkConf)
        
        val rdd1: RDD[Int] = sc.parallelize(List(1, 2, 3))
        val rdd2: RDD[Int] = sc.makeRDD(List(4, 5, 6))
        
        rdd1.collect().foreach(println)
        rdd2.collect().foreach(println)
        sc.stop();

默认情况下，`Spark`可以将一个作业切分多个任务后，发送给`Executor`节点并行计算，而能够并行计算的任务数量我们称之为并行度。这个数量可以在构建`RDD`时指定。记住，这里的并行执行的任务数量。

val rdd1: RDD[Int] = sc.parallelize(List(1, 2, 3), 4)

默认并行度为当前环境的最大`CPU`核数，可以指定并行度。

**② 从外部存储（文件）创建RDD**

由外部存储系统的数据集创建`RDD`包括：本地的文件系统，所有`Hadoop`支持的数据集，比如`HDFS`、`HBase`等。

val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDDTest2")
        val sc = new SparkContext(sparkConf)
        
    	/** path:读取文件的路径，path路径根据环境的不同自动发生改变，文件路径可以使用通配符，还可以使用第三方存储系统：HDFS */
        /** Spark读取文件时，默认采用的是Hadoop读取文件的方式：一行一行读 */	
        val rdd3: RDD[String] = sc.textFile("input")
        rdd3.collect().foreach(println)
    
        sc.stop();

**③ 从其他RDD创建**

主要是通过一个`RDD`运算完后，再产生新的`RDD`。详情请参考后续章节

**④ 直接创建RDD（new）**

使用`new`的方式直接构造`RDD`，一般由`Spark`框架自身使用。

[20200924102200408.png]: /images/20221123/2ff2c76145904d28b8adb36bced8298b.png
[20200924102249678.png]: /images/20221123/1ea06e6a57b048128993f2b89fa8de79.png
[20200924102313904.png]: /images/20221123/03d571fe4ed846a9bd5d0982631e8145.png
[20200924102404469.png]: /images/20221123/ee2385f775ed48fca2e875e49ffec1b7.png