Spark的三大数据结构-RDD创建

ゝ一世哀愁。 2023-10-04 16:56 24阅读 0赞

## Spark的三大数据结构-RDD创建 ##

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

### 一、特点 ###

1.  弹性  
    存储的弹性：内存与磁盘的自动切换；  
    容错的弹性：数据丢失可以自动恢复；  
    计算的弹性：计算出错重试机制；  
    分片的弹性：可根据需要重新分片。
2.  分布式：数据存储在大数据集群不同节点上
3.  数据集：RDD 封装了计算逻辑，并不保存数据
4.  数据抽象：RDD 是一个抽象类，需要子类具体实现
5.  不可变：RDD 封装了计算逻辑，是不可以改变的，想要改变，只能产生新的 RDD，在新的 RDD 里面封装计算逻辑
6.  可分区、并行计算

### 二、核心属性 ###

#### 1：分区列表 ####

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

#### 2：分区计算函数 ####

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

#### 3：RDD 之间的依赖关系 ####

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

#### 4：分区器（可选） ####

当数据为 KV 类型数据时，可以通过设定分区器自定义数据的分区

#### 5：首选位置（可选） ####

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

### 三、RDD的创建 ###

#### 1：从集合（内存）中创建 ####

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

// TODO 准备环境
    //"local[*]"代表本机的最大线程，如果不写，就是单线程
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark") 
    val sparkContext = new SparkContext(sparkConf) 
    
    // TODO 创建RDD
    // parallelize(): 并行，后面传入的是一个集合，这个集合就是需要处理的数据
    val rdd1 = sparkContext.parallelize( 
        List(1,2,3,4) 
    ) 
    // makeRDD语义上相较于parallelize()更容易理解，其实底端调用的就是parallelize()
    val rdd2 = sparkContext.makeRDD( 
        List(1,2,3,4) 
    ) 
    // 调用collect时才会执行
    rdd1.collect().foreach(println) 
    rdd2.collect().foreach(println) 
    
    //资源关闭
    sparkContext.stop()

**说明：**

1.  SparkConf().setMaster("local\[*\]").setAppName(“spark”) ："local\[*\]"代表本机的最大线程，如果不写，就是单线程
2.  sparkContext.parallelize(List)：parallelize(): 并行，后面传入的是一个集合，这个集合就是需要处理的数据
3.  sparkContext.makeRDD(List)：makeRDD语义上相较于parallelize()更容易理解，其实底端调用的就是parallelize()
4.  rdd1.collect().foreach(println) ：调用collect时才会执行

#### 2：从外部存储（文件）创建RDD ####

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

// TODO 准备环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark") 
    val sparkContext = new SparkContext(sparkConf) 
    
    //TODO 创建RDD
    //从文件创建RDD，将文件中的数据作为处理的数据源
    //path：可以使绝对路径，也可以是相对路径，相对路径的根节点就是本项目
    //path的路径可以是文件，也可以是文件夹
    //path路径还可以写通配符 sc.textFile("datas/1*.txt")
    //path还可以写分布式存储系统路径 sc.textFile("hdfs://hadoop102:8020/test.txt")
    val fileRDD: RDD[String] = sparkContext.textFile("input") 
    fileRDD.collect().foreach(println) 
    
    //TODO 资源关闭
    sparkContext.stop()

**说明：**

1.  从文件创建RDD，将文件中的数据作为处理的数据源
2.  path：可以使绝对路径，也可以是相对路径，相对路径的根节点就是本项目
3.  path的路径可以是文件，也可以是文件夹
4.  path路径还可以写通配符 sc.textFile(“datas/1\*.txt”)
5.  path还可以写分布式存储系统路径 sc.textFile(“hdfs://hadoop102:8020/test.txt”)