Spark RDD

淡淡的烟草味﹌ 2023-05-31 10:09 30阅读 0赞

## Spark RDD概述 ##

弹性分布式数据集 Resilient Distributed Datasets (RDDs)，是Spark中最基本的数据（计算）抽象。代码中是一个抽象类，它代表一个不可变、可分区、里面的元素可并行计算的集合。创建RDD的方法有两种：并行化 驱动程序中的现有集合，或引用外部存储系统（例如共享文件系统，HDFS，HBase或提供Hadoop InputFormat的任何数据源）中的数据集。

### 1. RDD特点： ###

*  **RDD逻辑上是分区的，一个 RDD 由一个或者多个分区（Partitions）组成。对于 RDD 来说，每个分区会被一个计算任务所处理，用户可以在创建 RDD 时指定其分区个数，如果没有指定，则默认采用程序所分配到的 CPU 的核心数。**
 *  **计算的时候会通过一个[compute函数][compute]计算每个分区的数据。**
 *  **RDD是只读的，要想改变RDD中的数据，只能在现有的RDD基础上创建新的RDD。**
 *  **RDDs通过操作算子进行转换，转换得到的新RDD包含了从其他RDDs衍生所必需的信息，RDDs之间维护着这种血缘关系，也称之为依赖。依赖包括两种，一种是窄依赖，RDDs之间分区是一一对应的，另一种是宽依赖，下游RDD的每个分区与上游RDD(也称之为父RDD)的每个分区都有关，是多对多的关系。**
 *  **RDD 会保存彼此间的依赖关系，RDD 的每次转换都会生成一个新的依赖关系，这种 RDD 之间的依赖关系就像流水线一样。在部分分区数据丢失后，可以通过这种依赖关系重新计算丢失的分区数据，而不是对 RDD 的所有分区进行重新计算；**
 *  **Key-Value 型的 RDD 还拥有 Partitioner(分区器)，用于决定数据被存储在哪个分区中，目前 Spark 中支持 HashPartitioner(按照哈希分区) 和 RangeParationer(按照范围进行分区)；**
 *  **一个优先位置列表 (可选)，用于存储每个分区的优先位置 (prefered location)。对于一个 HDFS 文件来说，这个列表保存的就是每个分区所在的块的位置，按照“移动数据不如移动计算“的理念，Spark 在进行任务调度的时候，会尽可能的将计算任务分配到其所要处理数据块的存储位置。**
 *  **如果在应用程序中多次使用同一个RDD，可以将该RDD缓存起来，该RDD只有在第一次计算的时候会根据血缘关系得到分区的数据，在后续其他地方用到该RDD的时候，会直接从缓存处取而不用再根据血缘关系计算，这样就加速后期的重用。**

`RDD[T]` 抽象类的部分相关代码如下：

@DeveloperApi
      def compute(split: Partition, context: TaskContext): Iterator[T]
      
    // 获取所有分区
    protected def getPartitions: Array[Partition]
    
    // 获取所有依赖关系
    protected def getDependencies: Seq[Dependency[_]] = deps
    
    // 获取优先位置列表
    protected def getPreferredLocations(split: Partition): Seq[String] = Nil
    
    // 分区器 由子类重写以指定它们的分区方式
    @transient val partitioner: Option[Partitioner] = None
    
    //缓存，默认是MEMORY_ONLY
    def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)

### 2.RDD的创建 ###

**Spark围绕弹性分布式数据集（RDD）的概念展开，RDD是可并行操作的元素的容错集合。创建RDD的方法有两种：并行化 驱动程序中的现有集合，或引用外部存储系统（例如共享文件系统，HDFS，HBase或提供Hadoop InputFormat的任何数据源）中的数据集**。

#### 1.通过在驱动程序创建 ####

通过在驱动程序（Scala ）中的现有集合上调用SparkContext的parallelize方法或makeRDD方法来创建并行集合Seq。复制集合的元素以形成可以并行操作的分布式数据集。例如，以下是创建包含数字1到5的并行化集合的方法：

val data = Array(1, 2, 3, 4, 5)
    val distData = sc.parallelize(data)

或

val data1 = Array(1, 2, 3, 4, 5)
    val distData1 = sc.makeRDD(data1)

如图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70]

#### 2.通过外部存储系统的数据集 ####

//本地文件系统
    val distFile = sc.textFile("file:///home/hadoop/user.json")
    //hdfs文件系统,我的系统默认使用该文件系统
    val distFile = sc.textFile("/user.json")

注意事项

1.  Spark的所有基于文件的输入方法（包括textFile）都支持在目录，压缩文件和通配符上运行。例如，你可以使用textFile("/my/directory")，textFile("/my/directory/*.txt")和textFile("/my/directory/*.gz")。
2.  该textFile方法还采用可选的第二个参数来控制**文件的分区数**。默认情况下，Spark为文件的每个块创建一个分区（HDFS中的块默认为128MB），但是您也可以通过传递更大的值来请求更大数量的分区。请注意，分区不能少于块。

### textFile & wholeTextFiles ###

两者都可以用来读取外部文件，但是返回格式是不同的：

*  **textFile**：其返回格式是 `RDD[String]` ，返回的是就是文件内容，RDD 中每一个元素对应一行数据；
 *  **wholeTextFiles**：其返回格式是 `RDD[(String, String)]`，元组中第一个参数是文件路径，第二个参数是文件内容；
 *  两者都提供第二个参数来控制最小分区数；
 *  从 HDFS 上读取文件时，Spark 会为每个块创建一个分区。

def textFile(path: String,minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {...}
    def wholeTextFiles(path: String,minPartitions: Int = defaultMinPartitions): RDD[(String, String)]={..}

### 3.RDD操作 ###

**RDD 支持两种类型的操作：*transformations*（转换，从现有数据集创建新数据集）和 *actions*（在数据集上运行计算后将值返回到驱动程序）。RDD 中的所有转换操作都是惰性的，它们只是记住这些转换操作，但不会立即执行，只有遇到 *action* 操作后才会真正的进行计算，这类似于函数式编程中的惰性求值。**

val list = List(1, 2, 3)
    // map 是一个 transformations 操作，而 foreach 是一个 actions 操作
    sc.parallelize(list).map(_ * 10).foreach(println)
    // 输出： 10 20 30

### 4.RDD缓存 ###

**Spark 速度非常快的一个原因是 RDD 支持缓存。成功缓存后，如果之后的操作使用到了该数据集，则直接从缓存中获取。虽然缓存也有丢失的风险，但是由于 RDD 之间的依赖关系，如果某个分区的缓存数据丢失，只需要重新计算该分区即可。**

RDD通过persist方法或cache方法可以将前面的计算结果缓存，默认情况下 persist() 会把数据以序列化的形式缓存在 JVM 的堆空间中，但是并不是这两个方法被调用时立即缓存，而是触发后面的action时，该RDD将会被缓存在计算节点的内存中，并供后面重用。RDD默认的存储级别是仅在内存存储一份，Spark的存储级别还有好多种，存储级别在object StorageLevel中定义的。

**缓存相关的代码如下：**

/** * Persist this RDD with the default storage level (`MEMORY_ONLY`). */
      def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)
    
      /** * Persist this RDD with the default storage level (`MEMORY_ONLY`). */
      def cache(): this.type = persist()

**缓存级别相关的代码**

val NONE = new StorageLevel(false, false, false, false)
      //只在磁盘上缓存 RDD
      val DISK_ONLY = new StorageLevel(true, false, false, false)
      //上面的对应级别功能相同，但是会为每个分区在集群中的两个节点上建立副本
      val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
      //默认的缓存级别，将 RDD 以反序列化的 Java 对象的形式存储在 JVM 中。如果内存空间不够，则部分分区数据将不再缓存
      val MEMORY_ONLY = new StorageLevel(false, true, false, true)
     // 与上面的对应级别功能相同，但是会为每个分区在集群中的两个节点上建立副本。
      val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
     //将 RDD 以序列化的 Java 对象的形式进行存储（每个分区为一个 byte 数组）。这种方式比反序列化对象节省存储空间，但在读取时会增加 CPU 的计算负担。仅支持 Java 和 Scala 
      val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
      val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
      //将 RDD 以反序列化的 Java 对象的形式存储 JVM 中。如果内存空间不够，将未缓存的分区数据存储到磁盘，在需要使用这些分区时从磁盘读取。
      val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
      val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
      //类似于 `MEMORY_ONLY_SER`，但是溢出的分区数据会存储到磁盘，而不是在用到它们时重新计算。仅支持 Java 和 Scala。
      val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
      val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
      //与 `MEMORY_ONLY_SER` 类似，但将数据存储在堆外内存中。这需要启用堆外内存。
      val OFF_HEAP = new StorageLevel(true, true, true, false, 1)

**移除缓存**

Spark 会自动监视每个节点上的缓存使用情况，并按照最近最少使用（LRU）的规则删除旧数据分区。当然，你也可以使用 `RDD.unpersist()` 方法进行手动删除。

### 5.RDD的依赖关系 ###

RDD 和它的父 RDD(s) 之间的依赖关系分为两种不同的类型：

*  **窄依赖 (narrow dependency)**：父 RDDs 的一个分区最多被子 RDDs 一个分区所依赖；
 *  **宽依赖 (wide dependency)**：父 RDDs 的一个分区可以被子 RDDs 的多个子分区所依赖。  
    用两张图来表示这两种依赖类型

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 2]  
区分这两种依赖是非常有用的：

*  首先，窄依赖允许在一个集群节点上以流水线的方式（pipeline）对父分区数据进行计算，例如先执行 map 操作，然后执行 filter 操作。而宽依赖则需要计算好所有父分区的数据，然后再在节点之间进行 Shuffle，这与 MapReduce 类似。
 *  窄依赖能够更有效地进行数据恢复，因为只需重新对丢失分区的父分区进行计算，且不同节点之间可以并行计算；而对于宽依赖而言，如果数据丢失，则需要对所有父分区数据进行计算并再次 Shuffle。

### 6.DAG的生成 ###

RDD(s) 及其之间的依赖关系组成了 DAG(有向无环图)，DAG 定义了这些 RDD(s) 之间的 Lineage(血统) 关系，通过血统关系，如果一个 RDD 的部分或者全部计算结果丢失了，也可以重新进行计算。那么 Spark 是如何根据 DAG 来生成计算任务呢？主要是根据依赖关系的不同将 DAG 划分为不同的计算阶段 (Stage)：

*  对于窄依赖，由于分区的依赖关系是确定的，其转换操作可以在同一个线程执行，所以可以划分到同一个执行阶段；
 *  对于宽依赖，由于 Shuffle 的存在，只能在父 RDD(s) 被 Shuffle 处理完成后，才能开始接下来的计算，因此遇到宽依赖就需要重新划分阶段。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 3]

spark官网[http://spark.apache.org/docs/latest/rdd-programming-guide.html\#resilient-distributed-datasets-rdds][http_spark.apache.org_docs_latest_rdd-programming-guide.html_resilient-distributed-datasets-rdds]  
spark分读取文件分区规则[https://blog.csdn.net/u014756380/article/details/78727386][https_blog.csdn.net_u014756380_article_details_78727386]

[compute]: https://blog.csdn.net/jiangpeng59/article/details/53213694
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70]: /images/20230531/dad3199a77cc43d4b15fdd50c5cd9462.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 1]: /images/20230531/945f03945ad642b5969038fa7db9206d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 2]: /images/20230531/084ecb64110549ad919d013c60137445.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 3]: /images/20230531/65fbc9671c11449e84e7b1932129a725.png
[http_spark.apache.org_docs_latest_rdd-programming-guide.html_resilient-distributed-datasets-rdds]: http://spark.apache.org/docs/latest/rdd-programming-guide.html#resilient-distributed-datasets-rdds
[https_blog.csdn.net_u014756380_article_details_78727386]: https://blog.csdn.net/u014756380/article/details/78727386