spark系列4：RDD介绍与使用

ゝ一世哀愁。 2022-11-09 14:13 121阅读 0赞

## 4. RDD 入门 ##

目标

上面通过一个 WordCount 案例, 演示了 Spark 大致的编程模型和运行方式, 接下来针对 Spark 的编程模型做更详细的扩展

1.  理解 WordCount 的代码
    
    1.  从执行角度上理解, 数据之间如何流转
    2.  从原理角度理解, 各个算子之间如何配合
2.  粗略理解 Spark 中的编程模型 RDD
3.  理解 Spark 中 RDD 的各个算子

object WordCounts {
    
      def main(args: Array[String]): Unit = {
        // 1. 创建 Spark Context
        val conf = new SparkConf().setMaster("local[2]")
        val sc: SparkContext = new SparkContext(conf)
    
        // 2. 读取文件并计算词频
        val source: RDD[String] = sc.textFile("hdfs://node01:8020/dataset/wordcount.txt", 2)
        val words: RDD[String] = source.flatMap { line => line.split(" ") }
        val wordsTuple: RDD[(String, Int)] = words.map { word => (word, 1) }
        val wordsCount: RDD[(String, Int)] = wordsTuple.reduceByKey { (x, y) => x + y }
    
        // 3. 查看执行结果
        println(wordsCount.collect)
      }
    }

在这份 WordCount 代码中, 大致的思路如下:

1.  使用 `sc.textFile()` 方法读取 HDFS 中的文件, 并生成一个 `RDD`
2.  使用 `flatMap` 算子将读取到的每一行字符串打散成单词, 并把每个单词变成新的行
3.  使用 `map` 算子将每个单词转换成 `(word, 1)` 这种元组形式
4.  使用 `reduceByKey` 统计单词对应的频率

其中所使用到的算子有如下几个:

*  `flatMap` 是一对多
 *  `map` 是一对一
 *  `reduceByKey` 是按照 Key 聚合, 类似 MapReduce 中的 Shuffled

如果用图形表示的话, 如下:

![2d5bc5474ac87123de26d9c5ca402dd4][]

总结以及引出新问题

上面大概说了两件事:

1.  代码流程
2.  算子

在代码中有一些东西并未交代:

1.  source, words, wordsTuple 这些变量的类型是 `RDD[Type]`, 什么是 `RDD`?
2.  还有更多算子吗?

RDD 是什么

![fa029f454c7b6445fa72ea6df999f67e][]

定义

RDD, 全称为 Resilient Distributed Datasets, 是一个容错的, 并行的数据结构, 可以让用户显式地将数据存储到磁盘和内存中, 并能控制数据的分区.

同时, RDD 还提供了一组丰富的操作来操作这些数据. 在这些操作中, 诸如 map, flatMap, filter 等转换操作实现了 Monad 模式, 很好地契合了 Scala 的集合操作. 除此之外, RDD 还提供了诸如 join, groupBy, reduceByKey 等更为方便的操作, 以支持常见的数据运算.

通常来讲, 针对数据处理有几种常见模型, 包括: Iterative Algorithms, Relational Queries, MapReduce, Stream Processing. 例如 Hadoop MapReduce 采用了 MapReduce 模型, Storm 则采用了 Stream Processing 模型. RDD 混合了这四种模型, 使得 Spark 可以应用于各种大数据处理场景.

RDD 作为数据结构, 本质上是一个只读的分区记录集合. 一个 RDD 可以包含多个分区, 每个分区就是一个 DataSet 片段.

RDD 之间可以相互依赖, 如果 RDD 的每个分区最多只能被一个子 RDD 的一个分区使用，则称之为窄依赖, 若被多个子 RDD 的分区依赖，则称之为宽依赖. 不同的操作依据其特性, 可能会产生不同的依赖. 例如 map 操作会产生窄依赖, 而 join 操作则产生宽依赖.

特点

1.  RDD 是一个编程模型
    
    1.  RDD 允许用户显式的指定数据存放在内存或者磁盘
    2.  RDD 是分布式的, 用户可以控制 RDD 的分区
2.  RDD 是一个编程模型
    
    1.  **RDD 提供了丰富的操作**
    2.  **RDD 提供了 map, flatMap, filter 等操作符, 用以实现 Monad 模式**
    3.  **RDD 提供了 reduceByKey, groupByKey 等操作符, 用以操作 Key-Value 型数据**
    4.  **RDD 提供了 max, min, mean 等操作符, 用以操作数字型的数据**
3.  RDD 是混合型的编程模型, 可以支持迭代计算, 关系查询, MapReduce, 流计算
4.  RDD 是只读的
5.  RDD 之间有依赖关系, 根据执行操作的操作符的不同, 依赖关系可以分为宽依赖和窄依赖

RDD 的分区

![f738dbe3df690bc0ba8f580a3e2d1112][]

整个 WordCount 案例的程序从结构上可以用上图表示, 分为两个大部分

存储

文件如果存放在 HDFS 上, 是分块的, 类似上图所示, 这个 `wordcount.txt` 分了三块

计算

Spark 不止可以读取 HDFS, Spark 还可以读取很多其它的数据集, Spark 可以从数据集中创建出 RDD

例如上图中, 使用了一个 RDD 表示 HDFS 上的某一个文件, 这个文件在 HDFS 中是分三块, 那么 RDD 在读取的时候就也有三个分区, 每个 RDD 的分区对应了一个 HDFS 的分块

后续 RDD 在计算的时候, 可以更改分区, 也可以保持三个分区, 每个分区之间有依赖关系, 例如说 RDD2 的分区一依赖了 RDD1 的分区一

RDD 之所以要设计为有分区的, 是因为要进行分布式计算, 每个不同的分区可以在不同的线程, 或者进程, 甚至节点中, 从而做到并行计算

总结

1.  RDD 是弹性分布式数据集
2.  RDD 一个非常重要的前提和基础是 RDD 运行在分布式环境下, 其可以分区

### 4.1. 创建 RDD ###

程序入口 SparkContext

val conf = new SparkConf().setMaster("local[2]")
    val sc: SparkContext = new SparkContext(conf)

`SparkContext` 是 spark-core 的入口组件, 是一个 Spark 程序的入口, 在 Spark 0.x 版本就已经存在 `SparkContext` 了, 是一个元老级的 API

如果把一个 Spark 程序分为前后端, 那么服务端就是可以运行 Spark 程序的集群, 而 `Driver` 就是 Spark 的前端, 在 `Driver` 中 `SparkContext` 是最主要的组件, 也是 `Driver` 在运行时首先会创建的组件, 是 `Driver` 的核心

`SparkContext` 从提供的 API 来看, 主要作用是连接集群, 创建 RDD, 累加器, 广播变量等

简略的说, RDD 有三种创建方式

*  RDD 可以通过本地集合直接创建
 *  RDD 也可以通过读取外部数据集来创建
 *  RDD 也可以通过其它的 RDD 衍生而来

通过本地集合直接创建 RDD

val conf = new SparkConf().setMaster("local[2]")
    val sc = new SparkContext(conf)
    
    val list = List(1, 2, 3, 4, 5, 6)
    val rddParallelize = sc.parallelize(list, 2)
    val rddMake = sc.makeRDD(list, 2)

通过 `parallelize` 和 `makeRDD` 这两个 API 可以通过本地集合创建 RDD

这两个 API 本质上是一样的, 在 `makeRDD` 这个方法的内部, 最终也是调用了 `parallelize。区别就是parallelize可以不指定分区数。`

因为不是从外部直接读取数据集的, 所以没有外部的分区可以借鉴, 于是在这两个方法都都有两个参数, 第一个参数是本地集合, 第二个参数是分区数

通过读取外部文件创建 RDD

val conf = new SparkConf().setMaster("local[2]")
    val sc = new SparkContext(conf)
    
    val source: RDD[String] = sc.textFile("hdfs://node01:8020/dataset/wordcount.txt")

*  访问方式
    
     *  支持访问文件夹, 例如 `sc.textFile("hdfs:///dataset")`
     *  支持访问压缩文件, 例如 `sc.textFile("hdfs:///dataset/words.gz")`
     *  支持通过通配符访问, 例如 `sc.textFile("hdfs:///dataset/*.txt")`

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td> <p>如果把 Spark 应用跑在集群上, 则 Worker 有可能在任何一个节点运行</p> <p>所以如果使用&nbsp;<code>file:///…;</code>&nbsp;形式访问本地文件的话, 要确保所有的 Worker 中对应路径上有这个文件, 否则可能会报错无法找到文件</p> </td> 
  </tr> 
 </tbody> 
</table>

*  **分区**
    
     *  默认情况下读取 HDFS 中文件的时候, 每个 HDFS 的 `block` 对应一个 RDD 的 `partition`, `block` 的默认是128M
     *  通过第二个参数, 可以指定分区数量, 例如 `sc.textFile("hdfs://node01:8020/dataset/wordcount.txt", 20)`
     *  如果通过第二个参数指定了分区, 这个分区数量一定不能小于\`block\`数

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td>通常每个 CPU core 对应 2 - 4 个分区是合理的值</td> 
  </tr> 
 </tbody> 
</table>

*  支持的平台
    
     *  支持 Hadoop 的几乎所有数据格式, 支持 HDFS 的访问
     *  通过第三方的支持, 可以访问AWS和阿里云中的文件, 详情查看对应平台的 API

通过其它的 RDD 衍生新的 RDD

val conf = new SparkConf().setMaster("local[2]")
    val sc = new SparkContext(conf)
    
    val source: RDD[String] = sc.textFile("hdfs://node01:8020/dataset/wordcount.txt", 20)
    val words = source.flatMap { line => line.split(" ") }

*  `source` 是通过读取 HDFS 中的文件所创建的
 *  `words` 是通过 `source` 调用算子 `map` 生成的新 RDD

总结

1.  RDD 的可以通过三种方式创建, 通过本地集合创建, 通过外部数据集创建, 通过其它的 RDD 衍生

### 4.2. RDD 算子 ###

目标

1.  理解各个算子的作用
2.  通过理解算子的作用, 反向理解 WordCount 程序, 以及 Spark 的要点

Map 算子

sc.parallelize(Seq(1, 2, 3))
      .map( num => num * 10 )
      .collect()

![c59d44296918b864a975ebbeb60d4c04][]

![57c2f77284bfa8f99ade091fdd7e9f83][]

作用

把 RDD 中的数据 一对一 的转为另一种形式

调用

`def map[U: ClassTag](f: T ⇒ U): RDD[U]`

参数

`f` → Map 算子是 `原RDD → 新RDD` 的过程, 这个函数的参数是原 RDD 数据, 返回值是经过函数转换的新 RDD 的数据

注意点

Map 是一对一, 如果函数是 `String → Array[String]` 则新的 RDD 中每条数据就是一个数组

FlatMap 算子

sc.parallelize(Seq("Hello lily", "Hello lucy", "Hello tim"))
      .flatMap( line => line.split(" ") )
      .collect()

![f6c4feba14bb71372aa0cb678067c6a8][]

![ec39594f30ca4d59e2ef5cdc60387866][]

作用

FlatMap 算子和 Map 算子类似, 但是 FlatMap 是一对多

调用

`def flatMap[U: ClassTag](f: T ⇒ List[U]): RDD[U]`

参数

`f` → 参数是原 RDD 数据, 返回值是经过函数转换的新 RDD 的数据, 需要注意的是返回值是一个集合, 集合中的数据会被展平后再放入新的 RDD

注意点

flatMap 其实是两个操作, 是 `map + flatten`, 也就是先转换, 后把转换而来的 List 展开

ReduceByKey 算子

sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
      .reduceByKey( (curr, agg) => curr + agg )
      .collect()

![07678e1b4d6ba1dfaf2f5df89489def4][]

![a9b444d144d6996c83b33f6a48806a1a][]

作用

首先按照 Key 分组, 接下来把整组的 Value 计算出一个聚合值, 这个操作非常类似于 MapReduce 中的 Reduce

调用

`def reduceByKey(func: (V, V) ⇒ V): RDD[(K, V)]`

参数

func → 执行数据处理的函数, 传入两个参数, 一个是当前值, 一个是局部汇总, 这个函数需要有一个输出, 输出就是这个 Key 的汇总结果

注意点

*  ReduceByKey 只能作用于 Key-Value 型数据, Key-Value 型数据在当前语境中特指 Tuple2
 *  ReduceByKey 是一个需要 Shuffled 的操作
 *  和其它的 Shuffled 相比, ReduceByKey是高效的, 因为类似 MapReduce 的, 在 Map 端有一个 Cominer, 这样 I/O 的数据便会减少

总结

1.  map 和 flatMap 算子都是转换, 只是 flatMap 在转换过后会再执行展开, 所以 map 是一对一, flatMap 是一对多
2.  reduceByKey 类似 MapReduce 中的 Reduce

[2d5bc5474ac87123de26d9c5ca402dd4]: /images/20221023/822fedbe3f1c46faa138c51046e60198.png
[fa029f454c7b6445fa72ea6df999f67e]: /images/20221023/b08f7856ba5a468198fe0527091b1e41.png
[f738dbe3df690bc0ba8f580a3e2d1112]: /images/20221023/d2c8e0c78e6d413cb01f6a3db5f58acf.png
[c59d44296918b864a975ebbeb60d4c04]: /images/20221023/313e4291b6a446cbaeb6a10a189646cc.png
[57c2f77284bfa8f99ade091fdd7e9f83]: /images/20221023/7d6e6d5f56504d60bc10df804d08489a.png
[f6c4feba14bb71372aa0cb678067c6a8]: /images/20221023/c487ed35e33048878b031592eaedfa47.png
[ec39594f30ca4d59e2ef5cdc60387866]: /images/20221023/14ef238db767407595ad3b7a49532ab3.png
[07678e1b4d6ba1dfaf2f5df89489def4]: /images/20221023/4f559ecc2b5a4a35aeee6d653157337c.png
[a9b444d144d6996c83b33f6a48806a1a]: /images/20221023/fc221da3c9e449618c8f0f02f2950366.png