SparkRDD简介

旧城等待， 2023-02-14 13:23 28阅读 0赞

参考文章：[spark的RDD中的action(执行)和transformation(转换)两种操作中常见函数介绍][spark_RDD_action_transformation]

[spark常见的RDD][spark_RDD]

spark的RDD中的action(执行)和transformation(转换)两种操作中常使用的函数

# 0. RDD产生背景 #

为了解决开发人员能在大规模的集群中以一种容错的方式进行内存计算，提出了 RDD 的概念，而当前的很多框架对迭代式算法场景与交互性数据挖掘场景的处理性能非常差, 这个是RDDs 的提出的动机。

# 1. RDD定义 #

RDD 是 Spark 的计算模型。RDD（Resilient Distributed Dataset）叫做弹性的分布式数据集合，是 Spark 中最基本的数据抽象，它代表一个不可变、只读的，被分区的数据集。操作 RDD 就像操作本地集合一样，有很多的方法可以调用，使用方便，而无需关心底层的调度细节。

弹性分布式数据集（RDD） 是spark为简化用户的使用，对所有的底层数据进行的抽象，以面向对象的方式提供了RDD很多的方法，通过这些方法来对RDD进行计算和输出。RDD是Spark的基石，所有的上层模块全部都是由RDD来实现。

Spark是以RDD概念为中心运行的。RDD是一个容错的、可以被并行操作的元素集合。创建一个RDD有两个方法：在你的驱动程序中并行化一个已经存在的集合；从外部存储系统中引用一个数据集。RDD的一大特性是分布式存储，分布式存储在最大的好处是可以让数据在不同工作节点并行存储，以便在需要数据时并行运算。弹性指其在节点存储时，既可以使用内存，也可已使用外存，为使用者进行大数据处理提供方便。除此之外，RDD的另一大特性是延迟计算，即一个完整的RDD运行任务被分为两部分：Transformation和Action.

在spark新版中，也许会有更多的action和transformation，可以参照spark的主页。  
hadoop提供的接口只有map和reduce函数，spark是mapreduce的扩展，提供两类操作，而不是两个，使使用更方便，开发时的代码量会尽量的被spark的这种多样的API减少数十倍。

# 2. RDD的特性 #

> a) 不可变， 对于所有RDD的操作都将产生一个新的RDD。  
> b) 可分区， RDD是通过将数据进行分区保存的。  
> c) 弹性  
>     存储的弹性：内存与磁盘的自动切换  
>     容错的弹性：数据丢失可以自动恢复  
>     计算的弹性：计算出错重试机制  
>     分片的弹性：根据需要重新分片

# 3. RDD分区  #

## 3.1 分区数量 ##

说对RDD进行操作 , 实际上是操作的RDD上的每一个分区 , 分区的数量决定了并行的数量 .  
使用 rdd.partitions.size 查看分区数量

scala> rdd.partitions.size
    res7: Int = 4
    
    scala> rdd2.partitions.size
    res8: Int = 4

## 3.2 初始化时分区的数量 ##

如果从外部创建RDD(比如从hdfs中读取数据)，正常情况下,

**分区的数量和读取的文件的block块数是一致的,**

但是如果只有一个block块,那么分区数量是2.也就是说最低的分区数量是2  
**如果是集合并行化创建得到的RDD，分区的数量，默认的和最大可用的cores数量相等。**  
**（--total-executor-cores > 可用的 cores? 可用的 cores:--total-executor-cores）**

**默认情况下，一个application使用多少个cores，就有多少个分区**  
分区的数量 = 运行任务的可用的cores（默认一个cores，能处理一个任务）

**正常情况下，读取HDFS中的文件，默认情况下，读到的文件有几个block块，得到的RDD就有几个分区。**

**当读取一个文件，不足一个block块的时候，会是2个分区**

**默认情况下，分区的数量  = 读取的文件的block块的数量，但是至少是2个**

scala> val rdd1 = sc.textFile("hdfs://hadoop01:9000/hbase-1.2.6-bin.tar.gz")
    scala> rdd1.partitions.size
    res2: Int = 1
    
    scala> val rdd2 = sc.textFile("hdfs://hadoop01:9000/hadoop-2.8.3.tar.gz")
    scala> rdd2.partitions.size
    res3: Int = 1
    
    scala> val rdd3 = sc.textFile("hdfs://hadoop01:9000/ideaIU-2017.2.2.exe")
    scala> rdd3.partitions.size
    res4: Int = 4

## 3.3 创建RDD时指定分区的数量 ##

通过集合并行化创建的RDD是可以任意修改分区的数量

val rdd = sc.makeRDD(arr,分区的数值)
    
    scala> val arr = Array(List(1,3),List(4,6))
    scala> val rdd3 = sc.parallelize(arr,3)
    scala> rdd3.partitions.size
    res1: Int = 3

# 4 RDD两种操作的简单介绍 #

## 4.1 Transformation ##

Transformation用于对RDD的创建，RDD只能使用Transformation创建，同时还提供大量操作方法，包括map，filter，groupBy，join等，RDD利用这些操作生成新的RDD，但是需要注意，无论多少次Transformation，在RDD中真正数据计算Action之前都不可能真正运行。

## 4.2 Action ##

Action是数据执行部分，其通过执行count，reduce，collect等方法真正执行数据的计算部分。实际上，RDD中所有的操作都是Lazy模式进行，运行在编译中不会立即计算最终结果，而是记住所有操作步骤和方法，只有显示的遇到启动命令才执行。这样做的好处在于大部分前期工作在Transformation时已经完成，当Action工作时，只需要利用全部自由完成业务的核心工作。

# 5.外部数据集到了spark经过哪些处理 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NodWltb2Zlbmd5YW5n_size_16_color_FFFFFF_t_70][]

# 6. RDD常见的一些转换算子 #

## 6.1 RDD的创建 ##

在Spark中创建RDD的创建方式大概可以分为三种：

### 6.1.1 从集合中创建RDD ###

从集合中创建RDD，Spark主要提供了两种函数：parallelize和makeRDD  
**a) 使用parallelize()从集合创建**

def parallelize[T: ClassTag]( seq: Seq[T], 
                     numSlices: Int = defaultParallelism): RDD[T]
    
    scala> val rdd = sc.parallelize(Array(1,2,3,4,5,6,7,8))
    rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] 
    at parallelize at <console>:24

**b) 使用makeRDD()从集合创建**

// 方法1:
    def makeRDD[T: ClassTag]( seq: Seq[T], numSlices: Int = defaultParallelism): 
    RDD[T] conf.getInt(“spark.default.parallelism”, math.max(totalCoreCount.get(), 2))
    
    // 方法2:
    // RDD[T] 可以为单个数据对象设置存放的节点
    def makeRDD[T: ClassTag](seq: Seq[(T, Seq[String])])
    
    // 示例:
    scala> val rdd1 = sc.makeRDD(Array(1,2,3,4,5,6,7,8)) 
    rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] 
          at makeRDD at <console>:24

### 6.1.2 从外部存储(如HDFS)创建RDD ###

// 读取HDFS文件系统
    val rdd2 = sc.textFile("hdfs://hadoop01:9000/words.txt")
    
    // 读取本地文件
    val rdd2 = sc.textFile(“file:///root/words.txt”)
    
    // 示例
    scala> val rdd2 = sc.textFile("file:root/word.txt")
    scala> rdd2.collect
    res2: Array[String] = Array(hadoop hbase java, hbase java spark, java, 
                                hadoop hive hive, hive hbase)
    
    scala> val rdd2= sc.textFile("hdfs://hadoop102:9000/RELEASE")
        rdd2: org.apache.spark.rdd.RDD[String] 
           = hdfs://hadoop102:9000/RELEASE     
    MapPartitionsRDD[4] at textFile at <console>:24

### 6.1.3 从其他RDD创建(**从父RDD转换成新的子RDD**) ###

调用 Transformation 类的方法，生成新的 RDD  
只要调用transformation类的算子，都会生成一个新的RDD。RDD中的数据类型，由传入给算子的函数的返回值类型决定  
注意：action类的算子，不会生成新的 RDD

scala> rdd.collect
    res3: Array[Int] = Array(1, 2, 3, 4, 5)
    
    scala> val rdd = sc.parallelize(arr)
    scala> val rdd2 = rdd.map(_*100)
    scala> rdd2.collect
    res4: Array[Int] = Array(100, 200, 300, 400, 500)

**Spark上的所有的方法 , 有一个专有的名词 , 叫做算子**

[spark_RDD_action_transformation]: https://blog.csdn.net/helloxiaozhe/article/details/78481784
[spark_RDD]: https://blog.csdn.net/shuimofengyang/article/details/100127655
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NodWltb2Zlbmd5YW5n_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190828222608802.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NodWltb2Zlbmd5YW5n,size_16,color_FFFFFF,t_70