SparkCore（四）：RDD基础编程

谁借莪１个温暖的怀抱￠ 2022-12-13 14:05 154阅读 0赞

### 文章目录 ###

*  一、RDD行动算子
 *   *  1.1 reduce
     *  1.2 collect
     *  1.3 count
     *  1.4 first
     *  1.5 take
     *  1.6 takeOrdered
     *  1.7 aggregate
     *  1.8 fold
     *  1.9 countByKey
     *  1.10 foreach
     *  1.11 save相关算子
 *  二、RDD序列化
 *  三、RDD依赖关系
 *  四、RDD持久化
 *  五、RDD文件读取与保存

# 一、RDD行动算子 #

> 行动算子不会产生新的RDD，而是触发作业的执行；行动算子执行后，会获取到作业的执行结果

## 1.1 reduce ##

函数签名：`def reduce(f: (T, T) => T): T`

函数说明：聚集`RDD`中的所有元素，先聚合分区内数据，再聚合分区间数据

val rdd = sc.makeRDD(List(1, 2, 3, 4))
        val result: Int = rdd.reduce(_ + _)
        println(result)//10

## 1.2 collect ##

函数签名：`def collect(): Array[T]`

函数说明：在驱动程序中，以数组`Array`的形式返回数据集的所有元素

val rdd = sc.makeRDD(List(1, 2, 3, 4))
        val result: Array[Int] = rdd.collect()
        result.foreach(t => print(t + " ")) //1 2 3 4

## 1.3 count ##

函数签名：`def count(): Long`

函数说明：返回`RDD`中元素的个数

val rdd = sc.makeRDD(List(1, 2, 3, 4))
        val count: Long = rdd.count()
        println(count) //4

## 1.4 first ##

函数签名：`def first(): T`

函数说明：返回`RDD`中元素的个数

val rdd = sc.makeRDD(List(1, 2, 3, 4))
        val first: Int = rdd.first()
        println(first) //1

## 1.5 take ##

函数签名：`def take(num: Int): Array[T]`

函数说明：返回一个由`RDD`的前`n`个元素组成的数组

val rdd = sc.makeRDD(List(1, 2, 3, 4))
        val takeArr: Array[Int] = rdd.take(2)
        takeArr.foreach(t => print(t + " "))//1 2

## 1.6 takeOrdered ##

函数签名：`def take(num: Int): Array[T]`

函数说明：返回该`RDD`排序后的前`n`个元素组成的数组

val rdd = sc.makeRDD(List(1, 3, 4, 2))
        // 默认升序
        val takeArr: Array[Int] = rdd.takeOrdered(2)((left, right) => { 
          right - left
        })
        takeArr.foreach(t => print(t + " "))//4 3

## 1.7 aggregate ##

函数签名：`def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U`

函数说明：分区的数据通过初始值和分区内的数据进行聚合，然后再和初始值进行分区间的数据聚合

val rdd = sc.makeRDD(List(1, 2, 3, 4))
        //第一个函数为分区内操作，第二个函数为分区间操作，第一个参数均为初始值
        val result: Int = rdd.aggregate(5)(_ + _, _ - _)
        println(result) // (1+2+3+4+5)-(5)=10

## 1.8 fold ##

函数签名：`def fold(zeroValue: T)(op: (T, T) => T): T`

函数说明：折叠操作，`aggregate`的简化版操作，分区内操作和分区间操作一样

val rdd = sc.makeRDD(List(1, 2, 3, 4))
        val result: Int = rdd.fold(5)(_ + _)
        println(result) //(1+2+3+4+5)+(5)=20

## 1.9 countByKey ##

函数签名：`def countByKey(): Map[K, Long]`

函数说明：统计每种`key`的个数

val rdd = sc.makeRDD(List(("a", 1), ("a", 1), ("b", 1)))
        val result: collection.Map[String, Long] = rdd.countByKey()
        result.foreach(t => print(t._1 + "=>" + t._2 + " "))//a=>2 b=>1

## 1.10 foreach ##

函数签名：

def foreach(f: T => Unit): Unit = withScope { 
        val cleanF = sc.clean(f)
        sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))
    }

函数说明：分布式遍历`RDD`中的每一个元素，调用指定函数

val rdd = sc.makeRDD(List(1, 2, 3, 4), 2)
        rdd.foreach(t => print(t + " "))//1 2 3 4

## 1.11 save相关算子 ##

函数签名：

def saveAsTextFile(path: String): Unit
    
    def saveAsObjectFile(path: String): Unit
    
    def saveAsSequenceFile(
      path: String,
      codec: Option[Class[_ <: CompressionCodec]] = None): Unit

函数说明：将数据保存到不同格式的文件中

// 保存成Text文件
    rdd.saveAsTextFile("output")
    
    // 序列化成对象保存到文件
    rdd.saveAsObjectFile("output1")
    
    // 保存成Sequencefile文件
    rdd.map((_,1)).saveAsSequenceFile("output2")

# 二、RDD序列化 #

**闭包检查**

从计算的角度, 算子以外的代码都是在`Driver`端执行, 算子里面的代码都是在`Executor`端执行。那么在`scala`的函数式编程中，就会导致算子内经常会用到算子外的数据，这样就形成了闭包的效果，如果使用的算子外的数据无法序列化，就意味着无法传值给`Executor`端执行，就会发生错误，所以需要在执行任务计算前，检测闭包内的对象是否可以进行序列化，这个操作我们称之为闭包检测。`Scala2.12`版本后闭包编译方式发生了改变

**Kryo序列化框架**

`Java`的序列化能够序列化任何的类。但是比较重（字节多），序列化后，对象的提交也比较大。`Spark`出于性能的考虑，`Spark2.0`开始支持另外一种`Kryo`序列化机制。`Kryo`速度是`Serializable`的10倍。当`RDD`在`Shuffle`数据的时候，简单数据类型、数组和字符串类型已经在`Spark`内部使用`Kryo`来序列化。

注意：即使使用`Kryo`序列化，也要继承`Serializable`接口

示例代码：

def main(args: Array[String]): Unit = { 
        val sparkConf = new SparkConf()
          .setMaster("local")
          .setAppName("SerializableTest2")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          .registerKryoClasses(Array(classOf[User]))
        val sc = new SparkContext(sparkConf)
        val rdd = sc.makeRDD(List(1, 2, 3, 4))
        val user = new User
        rdd.foreach(num => { 
          println("age = " + (user.age + num))
        })
      }
    
      class User extends Serializable { 
        val age: Int = 20
      }

# 三、RDD依赖关系 #

**RDD 血缘关系**

`RDD`只支持粗粒度转换，即在大量记录上执行的单个操作。将创建`RDD`的一系列`Lineage`（血统）记录下来，以便恢复丢失的分区。`RDD`的`Lineage`会记录`RDD`的元数据信息和转换行为，当该`RDD`的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区。

val linesRDD: RDD[String] = sc.textFile("input/1.txt")
        println(linesRDD.toDebugString)
        print("----------------")
        
        val wordsRDD: RDD[String] = linesRDD.flatMap(_.split(" "))
        println(wordsRDD.toDebugString)
        print("----------------")
        
        val mapRDD: RDD[(String, Int)] = wordsRDD.map((_, 1))
        println(mapRDD.toDebugString)
        print("----------------")
        
        val resultRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)
        println(resultRDD.toDebugString)
    
        resultRDD.collect()

打印：

(1) input/1.txt MapPartitionsRDD[1] at textFile at DependencyTest.scala:13 []
     |  input/1.txt HadoopRDD[0] at textFile at DependencyTest.scala:13 []
    ----------------
    (1) MapPartitionsRDD[2] at flatMap at DependencyTest.scala:17 []
     |  input/1.txt MapPartitionsRDD[1] at textFile at DependencyTest.scala:13 []
     |  input/1.txt HadoopRDD[0] at textFile at DependencyTest.scala:13 []
    ----------------
    (1) MapPartitionsRDD[3] at map at DependencyTest.scala:21 []
     |  MapPartitionsRDD[2] at flatMap at DependencyTest.scala:17 []
     |  input/1.txt MapPartitionsRDD[1] at textFile at DependencyTest.scala:13 []
     |  input/1.txt HadoopRDD[0] at textFile at DependencyTest.scala:13 []
    ----------------
    (1) ShuffledRDD[4] at reduceByKey at DependencyTest.scala:25 []
     +-(1) MapPartitionsRDD[3] at map at DependencyTest.scala:21 []
        |  MapPartitionsRDD[2] at flatMap at DependencyTest.scala:17 []
        |  input/1.txt MapPartitionsRDD[1] at textFile at DependencyTest.scala:13 []
        |  input/1.txt HadoopRDD[0] at textFile at DependencyTest.scala:13 []

**RDD 依赖关系**

这里所谓的依赖关系，其实就是`RDD`之间的关系

val linesRDD: RDD[String] = sc.textFile("input/1.txt")
        println(linesRDD.dependencies)
        println("----------------")
    
        val wordsRDD: RDD[String] = linesRDD.flatMap(_.split(" "))
        println(wordsRDD.dependencies)
        println("----------------")
    
        val mapRDD: RDD[(String, Int)] = wordsRDD.map((_, 1))
        println(mapRDD.dependencies)
        println("----------------")
    
        val resultRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)
        println(resultRDD.dependencies)
    
        resultRDD.collect()

打印：

List(org.apache.spark.OneToOneDependency@23940f86)
    ----------------
    List(org.apache.spark.OneToOneDependency@3b95d13c)
    ----------------
    List(org.apache.spark.OneToOneDependency@23f72d88)
    ----------------
    List(org.apache.spark.ShuffleDependency@6917bb4)

**RDD 窄依赖**

窄依赖表示每一个父`RDD`的`Partition`最多被子`RDD`的一个`Partition`使用，窄依赖我们形象的比喻为独生子女。

class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd)

**RDD 宽依赖**

宽依赖表示同一个父`RDD`的`Partition`被多个子`RDD`的`Partition`依赖，会引起`Shuffle`，总结：宽依赖我们形象的比喻为超生。

class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
        @transient private val _rdd: RDD[_ <: Product2[K, V]],
        val partitioner: Partitioner,
        val serializer: Serializer = SparkEnv.get.serializer,
        val keyOrdering: Option[Ordering[K]] = None,
        val aggregator: Option[Aggregator[K, V, C]] = None,
        val mapSideCombine: Boolean = false)
      extends Dependency[Product2[K, V]]

**RDD 阶段划分**

如果执行过程中中没有落盘的操作，那么应该是一个完整的阶段；如果执行过程中有落盘的操作，那么应该将阶段一分为二。

该`DAG`记录了`RDD`的转换过程和任务的阶段:  
![在这里插入图片描述][20201014172819395.png]  
**RDD 任务划分**

`RDD`任务切分中间分为：`Application`、`Job`、`Stage`和`Task`

*  `Application`：初始化一个`SparkContext`即生成一个`Application`；
 *  `Job`：一个`Action`算子就会生成一个`Job`；
 *  `Stage`：`Stage`等于宽依赖(`ShuffleDependency`)的个数加1；
 *  `Task`：一个`Stage`阶段中，最后一个`RDD`的分区个数就是`Task`的个数。

注意：`Application`\->`Job`\->`Stage`\->`Task`每一层都是1对`n`的关系。

# 四、RDD持久化 #

> 将计算结果进行缓存，重复使用，提高效率

**① RDD Cache缓存**

`RDD`通过`Cache`或者`Persist`方法将前面的计算结果缓存，默认情况下会把数据以序列化的形式缓存在`JVM`的堆内存中。但是并不是这两个方法被调用时立即缓存，而是触发后面的`action`算子时，该`RDD`将会被缓存在计算节点的内存中，并供后面重用。

// cache操作会增加血缘关系，不改变原有的血缘关系
    println(wordToOneRdd.toDebugString)
    
    // 数据缓存。
    wordToOneRdd.cache()
    
    // 可以更改存储级别
    //mapRdd.persist(StorageLevel.MEMORY_AND_DISK_2)

存储级别：

object StorageLevel { 
      val NONE = new StorageLevel(false, false, false, false)
      val DISK_ONLY = new StorageLevel(true, false, false, false)
      val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
      val MEMORY_ONLY = new StorageLevel(false, true, false, true)
      val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
      val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
      val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
      val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
      val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
      val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
      val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
      val OFF_HEAP = new StorageLevel(true, true, true, false, 1)

![在这里插入图片描述][20201020204058607.png]  
缓存有可能丢失，或者存储于内存的数据由于内存不足而被删除，`RDD`的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于`RDD`的一系列转换，丢失的数据会被重算，由于`RDD`的各个`Partition`是相对独立的，因此只需要计算丢失的部分即可，并不需要重算全部`Partition`。

`Spark`会自动对一些`Shuffle`操作的中间数据做持久化操作(比如：`reduceByKey`)。这样做的目的是为了当一个节点`Shuffle`失败了避免重新计算整个输入。但是，在实际使用的时候，如果想重用数据，仍然建议调用`persist`或`cache`。

**② RDD CheckPoint检查点**

所谓的检查点其实就是通过将`RDD`中间结果写入磁盘；

由于血缘依赖过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果检查点之后有节点出现问题，可以从检查点开始重做血缘，减少了开销。

对`RDD`进行`checkpoint`操作并不会马上被执行，必须执行`Action`操作才能触发。

sc.setCheckpointDir("./checkpoint1")
    
    // 创建一个RDD，读取指定位置文件:hello atguigu atguigu
    val lineRdd: RDD[String] = sc.textFile("input/1.txt")
    
    // 业务逻辑
    val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
    
    val wordToOneRdd: RDD[(String, Long)] = wordRdd.map { 
        word => { 
            (word, System.currentTimeMillis())
        }
    }
    
    // 增加缓存,避免再重新跑一个job做checkpoint
    wordToOneRdd.cache()
    // 数据检查点：针对wordToOneRdd做检查点计算
    wordToOneRdd.checkpoint()
    
    // 触发执行逻辑
    wordToOneRdd.collect().foreach(println)

**③ 缓存和检查点区别**

1.  `Cache`缓存只是将数据保存起来，不切断血缘依赖。`Checkpoint`检查点切断血缘依赖。
2.  `Cache`缓存的数据通常存储在磁盘、内存等地方，可靠性低。`Checkpoint`的数据通常存储在`HDFS`等容错、高可用的文件系统，可靠性高。
3.  建议对`Checkpoint`的`RDD`使用`Cache`缓存，这样`Checkpoint`的job只需从`Cache`缓存中读取数据即可，否则需要再从头计算一次`RDD`。

# 五、RDD文件读取与保存 #

`Spark`的数据读取及数据保存可以从两个维度来作区分：文件格式以及文件系统。

*  文件格式分为：`text`文件、`csv`文件、`sequence`文件以及`Object`文件；
 *  文件系统分为：本地文件系统、`HDFS`、`HBase`以及数据库。

**text 文件**

// 读取输入文件
    val inputRDD: RDD[String] = sc.textFile("input/1.txt")
    
    // 保存数据
    inputRDD.saveAsTextFile("output")

**sequence 文件**

`sequenceFile`文件是`Hadoop`用来存储二进制形式的`key-value`对而设计的一种平面文件(`Flat File`)。在`SparkContext`中，可以调用`sequenceFile[keyClass, valueClass](path)`。

// 保存数据为SequenceFile
    dataRDD.saveAsSequenceFile("output")
    
    // 读取SequenceFile文件
    sc.sequenceFile[Int,Int]("output").collect().foreach(println)

**object 对象文件**

对象文件是将对象序列化后保存的文件，采用`Java`的序列化机制。可以通过`objectFile[T: ClassTag](path)`函数接收一个路径，读取对象文件，返回对应的RDD，也可以通过调用`saveAsObjectFile()`实现对对象文件的输出。因为是序列化所以要指定类型。

// 保存数据
    dataRDD.saveAsObjectFile("output")
    
    // 读取数据
    sc.objectFile[Int]("output").collect().foreach(println)

[20201014172819395.png]: /images/20221123/798cc1d83cf440b897fd60cd4f2ab4d5.png
[20201020204058607.png]: /images/20221123/00e146544966463884db510f0e9888c1.png