Spark RDD详解！

柔情只为你懂 2022-12-08 12:49 176阅读 0赞

### 目录 ###

*   *  一、RDD基础
     *  二、创建RDD
     *  三、RDD操作
     *   *  （1）转化操作
         *  （2）行动操作
         *  （3）惰性求值
     *  四、常见的转化操作和行动操作
     *   *  （1）针对各元素的转化操作
         *  （2）伪集合操作
         *  （3）行动操作
     *  五、持久化（缓存）

Spark 对数据的核心抽象——弹性分布式数据集（Resilient Distributed Dataset， 简 称 RDD） 。 RDD 其实就是分布式的元素集合。在 Spark 中， 对数据的所有操作不外乎 **创 建 RDD**、 **转化已有 RDD** 以及 **调用 RDD 操作进行求值**。而在这一切背后， Spark 会自动将 RDD 中的数据分发到集群上，并将操作并行化执行。

## 一、RDD基础 ##

Spark 中的 RDD 就是一个不可变的分布式对象集合。每个 RDD 都被分为多个 分区 ，这些 分区运行在集群中的不同节点上。RDD 可以包含 Python、Java、Scala 中任意类型的对象， 甚至可以包含用户自定义的对象。

**用户可以使用两种方法创建 RDD：读取一个外部数据集， 或在驱动器程序里分发驱动器程 序中的对象集合（比如 list 和 set）**

创 建 出 来 后， RDD 支 持 两 种 类 型 的 操 作：  
**转 化 操 作 （transformation） 和 行 动 操 作（action）**

*  转化操作会由一个 RDD 生成一个新的 RDD  
    例如：根据谓词匹配情况筛选数 据就是一个常见的转化操作。
 *  行动操作 会对 RDD 计算出一个结果，并把结果返回到驱动器程序中，或把结 果存储到外部存储系统（如 HDFS）中  
    例如：first() 就是调用的一个行动操作， 它会返回 RDD 的第一个元素

转化操作和行动操作的区别：  
**转化操作和行动操作的区别在于 Spark 计算 RDD 的方式不同。 虽然你可以在任何时候定 义新的 RDD， 但 Spark 只会 惰性 计算这些 RDD。 它们只有第一次在一个行动操作中用到 时，才会真正计算。**

原因解释：  
我们以一个文本文件定义了数据， 然后把其中包 含 Python 的行筛选出来。如果 Spark 在我们运行 lines = sc.textFile(…) 时就把文件中 所有的行都读取并存储起来，就会消耗很多存储空间，而我们马上就要筛选掉其中的很多 数据。相反，一旦 Spark 了解了完整的转化操作链之后，它就可以只计算求结果时真正需 要的数据。 事实上， 在行动操作 first() 中， Spark 只需要扫描文件直到找到第一个匹配 的行为止，而不需要读取整个文件。

关于spark RDD持久化的解释：  
在默认情况下，Spark 的 RDD 会在你每次对它们进行行动操作时重新计算。如果想 在多个行动操作中重用同一个 RDD， 可以使用 RDD.persist() 让 Spark 把这个 RDD 缓存 下来。在第一次对持久化的 RDD 计算之后， Spark 会把 RDD 的内容保存到内存中（以分区方式 存储到集群中的各机器上），这样在之后的行动操作中，就可以重用这些数据了。我们也 可以把 RDD 缓存到磁盘上而不是内存中。不过这对于大规模数据集是很有意义的：如果不会重用该 RDD， 我们就没有必要浪费存储空 间， Spark 可以直接遍历一遍数据然后计算出结果。

总的来说，每个 Spark 程序或 shell 会话都按如下方式工作：

*  从外部数据创建出输入 RDD。
 *  使用诸如 ﬁlter() 这样的转化操作对 RDD 进行转化，以定义新的 RDD。
 *  告诉 Spark 对需要被重用的中间结果 RDD 执行 persist() 操作。
 *  使用行动操作（例如 count() 和 ﬁrst() 等）来触发一次并行计算， Spark 会对计算进行 优化后再执行。

## 二、创建RDD ##

**Spark 提供了两种创建 RDD 的方式：  
1.读取外部数据集  
2.在驱动器程序中对一个集合进行并行化。**

创建 RDD 最简单的方式就是把程序中一个已有的集合传给 SparkContext 的 parallelize() 方法，示例如下：

scala> val lines = sc.parallelize(List("pandas", "i like pandas"))
    lines: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at <console>:24

更常用的方式是从外部存储中读取数据来创建 RDD，示例如下：

scala> val rdd = sc.textFile("file:///opt/datas/stu.txt")
    rdd: org.apache.spark.rdd.RDD[String] = file:///opt/datas/stu.txt MapPartitionsRDD[3] at textFile at <console>:24

## 三、RDD操作 ##

RDD 支持两种操作： 转化操作 和 行动操作 。 RDD 的转化操作是返回一个新的 RDD 的操作， 比如 map() 和 filter() ，而行动操作则是向驱动器程序返回结果或 把结果写入外部系统的操作， 会触发实际的计算， 比如 count() 和 first() 。转化操作返回的是 RDD， 而行动操作返回的是其他的数据类型。

### （1）转化操作 ###

RDD 的转化操作是返回新 RDD 的操作。转化出来的 RDD 是惰性 求值的，只有在行动操作中用到这些 RDD 时才会被计算。许多转化操作都是针对各个元 素的，也就是说，这些转化操作每次只会操作 RDD 中的一个元素。不过并不是所有的转 化操作都是这样的。  
举个例子，假定我们有一个日志文件 log.txt， 内含有若干消息，希望选出其中的错误消息。 我们可以使用前面说过的转化操作 filter() 。

val inputRDD = sc.textFile("log.txt") 
    val errorsRDD = inputRDD.filter(line => line.contains("error"))

注意， filter() 操作不会改变已有的 inputRDD 中的数据。实际上，该操作会返回一个全新 的 RDD。 inputRDD 在后面的程序中还可以继续使用。比如我们还可以从中搜索别的单词。 事实上，要再从 inputRDD 中找出所有包含单词 warning 的行。接下来，我们使用另一个转 化操作 union() 来打印出包含 error 或 warning 的行数。 下例中用 Python 作了示例：

errorsRDD = inputRDD.filter(lambda x: "error" in x) 
    warningsRDD = inputRDD.filter(lambda x: "warning" in x) 
    badLinesRDD = errorsRDD.union(warningsRDD)

union() 与 filter() 的不同点在于它操作两个 数量的输入 RDD而不是一个。 转化操作可以操作任意数量的输入RDD。  
通过转化操作， 你从已有的 RDD 中派生出新的 RDD，Spark 会使用 谱系 图 （lineage graph）来记录这些不同 RDD 之间的依赖关系。 Spark 需要用这些信息来按需 计算每个 RDD， 也可以依靠谱系图在持久化的 RDD 丢失部分数据时恢复所丢失的数据。如图展示上述例子谱系图。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]

### （2）行动操作 ###

行动操作是第二种类型的 RDD 操作，它们会把最终求得的结 果返回到驱动器程序，或者写入外部存储系统中。由于行动操作需要生成实际的输出，它 们会强制执行那些求值必须用到的 RDD 的转化操作。

继续看上面的例子， 我们可能想输出关于 badLinesRDD 的一些信息。 为此， 需要使用两个行动操作来实现：用 count() 来返回计数结果， 用 take() 来收集 RDD 中的一些元素：  
Scala写法

println("Input had " + badLinesRDD.count() + " concerning lines")
    println("Here are 10 examples:") 
    badLinesRDD.take(10).foreach(println)

在这个例子中，我们在驱动器程序中使用 take() 获取了 RDD 中的少量元素。然后在本地 遍历这些元素，并在驱动器端打印出来。 RDD 还有一个 collect() 函数，可以用来获取整 个 RDD 中的数据。 如果你的程序把 RDD 筛选到一个很小的规模， 并且你想在本地处理 这些数据时，就可以使用它。记住，只有当你的整个数据集能在单台机器的内存中放得下 时，才能使用 collect() ，因此， collect() 不能用在大规模数据集上。  
在大多数情况下， RDD 不能通过 collect() 收集到驱动器进程中，因为它们一般都很大。 此时，我们通常要把数据写到诸如 HDFS 或 Amazon S3 这样的分布式的存储系统中。你可 以使用 saveAsTextFile() 、 saveAsSequenceFile() ， 或者任意的其他行动操作来把 RDD 的 数据内容以各种自带的格式保存起来。

**需要注意的是，每当我们调用一个新的行动操作时，整个 RDD 都会从头开始计算。要避 免这种低效的行为，用户可以将中间结果持久化**

### （3）惰性求值 ###

RDD 的转化操作都是惰性求值的。这意味着在被调用行动操作之前 Spark 不会 开始计算。惰性求值意味着当我们对 RDD 调用转化操作（例如调用 map() ）时，操作不会立即执行。 相反， Spark 会在内部记录下所要求执行的操作的相关信息。我们不应该把 RDD 看作存 放着特定数据的数据集，而最好把每个 RDD 当作我们通过转化操作构建出来的、记录如 何计算数据的指令列表。把数据读取到 RDD 的操作也同样是惰性的。因此，当我们调用 sc.textFile() 时，数据并没有读取进来，而是在必要时才会读取。和转化操作一样的是， 读取数据的操作也有可能会多次执行。

虽然转化操作是惰性求值的，但还是可以随时通过运行一个行动操作来强制 Spark 执行 RDD 的转化操作，比如使用 count() 。这是一种对你所写的程序 进行部分测试的简单方法。

Spark 使用惰性求值，这样就可以把一些操作合并到一起来减少计算数据的步骤。在类似 Hadoop MapReduce 的系统中，开发者常常花费大量时间考虑如何把操作组合到一起，以 减少 MapReduce 的周期数。而在 Spark 中，写出一个非常复杂的映射并不见得能比使用很 多简单的连续操作获得好很多的性能。因此，用户可以用更小的操作来组织他们的程序， 这样也使这些操作更容易管理。

## 四、常见的转化操作和行动操作 ##

这里我们会接触 Spark 中大部分常见的转化操作和行动操作。 包含特定数据类型的 RDD 还支持一些附加操作。

### （1）针对各元素的转化操作 ###

你很可能会用到的两个最常用的转化操作是 map() 和 filter() 。转化操作 map() 接收一个函数， 把这个函数用于 RDD 中的每个元素， 将函数的返回结果作为结果RDD 中对应元素的值。而转化操作filter()则接收一个函数，并将 RDD 中满足该函数的元素放入新的 RDD 中返回。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]

我们可以使用 map() 来做各种各样的事情：可以把我们的 URL 集合中的每个 URL 对应的 主机名提取出来，也可以简单到只对各个数字求平方值。 map() 的返回值类型不需要和输 入类型一样。 这样如果有一个字符串 RDD， 并且我们的 map() 函数是用来把字符串解析 并返回一个 Double 值的， 那么此时我们的输入 RDD 类型就是 RDD\[String\] ， 而输出类型 是 RDD\[Double\] 。

测试用 map() 对 RDD 中的所有数求平方：

scala> val input = sc.parallelize(List(1,2,3,4))
    input: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:24
    
    scala> val result = input.map(x => x*x)
    result: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[5] at map at <console>:25
    
    scala> println(result.collect().mkString(","))
    1,4,9,16

有时候，我们希望对每个输入元素生成多个输出元素。实现该功能的操作叫作 flatMap() 。 和 map() 类似，我们提供给 flatMap() 的函数被分别应用到了输入 RDD 的每个元素上。不 过返回的不是一个元素，而是一个返回值序列的迭代器。输出的 RDD 倒不是由迭代器组 成的。我们得到的是一个包含各个迭代器可访问的所有元素的 RDD。 flatMap() 的一个简 单用途是把输入的字符串切分为单词：

scala> val lines = sc.parallelize(List("hello world","hi","java python hadoop","c c++","hello me"))
    lines: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[6] at parallelize at <console>:24
    
    scala> val words = lines.flatMap(line => line.split(" "))
    words: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[7] at flatMap at <console>:25
    
    scala> words.count
    res10: Long = 10
    
    scala> words.first()
    res7: String = hello

图解RDD 的ﬂatMap()和map()的区别：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 2]

### （2）伪集合操作 ###

尽管 RDD 本身不是严格意义上的集合，但它也支持许多数学上的集合操作，比如合并和相 交操作。图中展示了四种操作。注意，这些操作都要求操作的 RDD 是相同数据类型的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 3]  
我们的 RDD 中最常缺失的集合属性是元素的唯一性， 因为常常有重复的元素。 如果只 要唯一的元素， 我们可以使用 RDD.distinct() 转化操作来生成一个只包含不同元素的新 RDD。 不过需要注意， distinct() 操作的开销很大，因为它需要将所有数据通过网络进行 混洗（shufﬂe），以确保每个元素都只有一份。

最简单的集合操作是 union(other) ，它会返回一个包含两个 RDD 中所有元素的 RDD。 这 在很多用例下都很有用，比如处理来自多个数据源的日志文件。与数学中的 union() 操作 不同的是，如果输入的 RDD 中有重复数据， Spark 的 union() 操作也会包含这些重复数据 （如有必要，我们可以通过 distinct() 实现相同的效果）。

Spark 还提供了 intersection(other) 方法，只返回两个 RDD 中都有的元素。 intersection() 在运行时也会去掉所有重复的元素（单个 RDD 内的重复元素也会一起移除）。 尽管 intersection() 与 union() 的概念相似， intersection() 的性能却要差很多， 因为它需要 通过网络混洗数据来发现共有的元素。

有时我们需要移除一些数据。 subtract(other) 函数接收另一个 RDD 作为参数， 返回 一个由只存在于第一个 RDD 中而不存在于第二个 RDD 中的所有元素组成的 RDD。 和 intersection() 一样，它也需要数据混洗。

我们也可以计算两个 RDD 的笛卡儿积，如图所示。 cartesian(other) 转化操作会返回 所有可能的 (a, b) 对，其中 a 是源 RDD 中的元素，而 b 则来自另一个 RDD。 笛卡儿积在 我们希望考虑所有可能的组合的相似度时比较有用，比如计算各用户对各种产品的预期兴 趣程度。我们也可以求一个 RDD 与其自身的笛卡儿积，这可以用于求用户相似度的应用 中。不过要特别注意的是，求大规模 RDD 的笛卡儿积开销巨大。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 4]  
**对一个数据为\{1, 2, 3, 3\}的RDD进行基本的RDD转化操作**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 5]  
**对数据分别为\{1, 2, 3\}和\{3, 4, 5\}的RDD进行针对两个RDD的转化操作**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 6]

### （3）行动操作 ###

**对一个数据为\{1, 2, 3, 3\}的RDD进行基本的RDD行动操作**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 7]

## 五、持久化（缓存） ##

如前所述， Spark RDD 是惰性求值的，而有时我们希望能多次使用同一个 RDD。 如果简单 地对 RDD 调用行动操作， Spark 每次都会重算 RDD 以及它的所有依赖。这在迭代算法中 消耗格外大，因为迭代算法常常会多次使用同一组数据。下例就是先对 RDD 作一次计 数、再把该 RDD 输出的一个小例子。  
Scala中的两次执行：

scala> val a = sc.parallelize(List(1,2,3,3))
    a: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[22] at parallelize at <console>:24
    scala> val result = a.map(x => (x*x))
    result: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[23] at map at <console>:25
    scala> println(result.count)
    4
    scala> result.collect
    res36: Array[Int] = Array(1, 4, 9, 9)

为了避免多次计算同一个 RDD， 可以让 Spark 对数据进行持久化。当我们让 Spark 持久化 存储一个 RDD 时，计算出 RDD 的节点会分别保存它们所求出的分区数据。如果一个有持 久化数据的节点发生故障， Spark 会在需要用到缓存的数据时重算丢失的数据分区。如果 希望节点故障的情况不会拖累我们的执行速度，也可以把数据备份到多个节点上。

出于不同的目的，我们可以为 RDD 选择不同的持久化级别（如图）。在 Scala和 Java 中， 默认情况下 persist() 会把数据以序列化的形式缓存在 JVM 的堆空 间中。在 Python 中，我们会始终序列化要持久化存储的数据，所以持久化级别默认值就是 以序列化后的对象存储在 JVM 堆空间中。当我们把数据写到磁盘或者堆外存储上时，也 总是使用序列化后的数据。

org.apache.spark.storage.StorageLevel 和 pyspark.StorageLevel 中的持久化级 别；如有必要，可以通过在存储级别的末尾加上“\_2”来把持久化数据存为两份  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 8]  
在 Scala 中使用 persist()：

scala> val a = sc.parallelize(List(1,2,3,3))
    a: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[22] at parallelize at <console>:24
    scala> val result = a.map(x => (x*x))
    result: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[23] at map at <console>:25
    scala> result.persist(org.apache.spark.storage.StorageLevel.DISK_ONLY)
    res40: result.type = MapPartitionsRDD[24] at map at <console>:25
    scala> result.collect
    res41: Array[Int] = Array(1, 4, 9, 9)
    scala> println(result.count)
    4

**注意，我们在第一次对这个RDD调用行动操作前就调用了 persist() 方法。persist() 调用本身不会触发强制求值。**

如果要缓存的数据太多，内存中放不下， Spark 会自动利用最近最少使用（LRU）的缓存 策略把最老的分区从内存中移除。对于仅把数据存放在内存中的缓存级别，下一次要用到 已经被移除的分区时，这些分区就需要重新计算。但是对于使用内存与磁盘的缓存级别的 分区来说，被移除的分区都会写入磁盘。不论哪一种情况，都不必担心你的作业因为缓存 了太多数据而被打断。不过，缓存不必要的数据会导致有用的数据被移出内存，带来更多重算的时间开销。

最后， RDD 还有一个方法叫作 unpersist() ， 调用该方法可以手动把持久化的 RDD 从缓 存中移除。

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/e495a71ed1de408e8223f8f22ed423d8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/bfe06eb70014407a86563a778e0cf6d0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/844e0e815034486c9077048d1bd127de.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/3f1fefc10101435c8fce63723f79f564.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/4e3fdf6684154242b138d69c3bbd93cf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/1d0eab95e7b542258b220c5571c4174f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/9e1e75c1e00b4727a61ef9a76dcc41ac.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221123/54c3834d294948fab6ff2a368f1276e6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221123/6cdcdcad40f046a38c75290f53d2c8fa.png