Spark-Core

雨点打透心脏的1/2处 2024-02-21 11:50 57阅读 0赞

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

[Spark简介][Spark]  
[Spark-Core核心算子][Spark-Core]  
[Spark-Core][Spark-Core 1]

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

#### 文章目录 ####

*   *  一、RDD 编程
     *   *  1、RDD序列化
         *   *  1.2 Kryo序列化框架
         *  2、RDD依赖关系
         *   *  2.1 查看血缘关系
             *  2.2 查看依赖关系
             *  2.3 窄依赖
             *  2.4 宽依赖
             *  2.5 Stage任务划分
         *  3、RDD 持久化
         *   *  3.1 Cache缓存
             *  3.2 CheckPoint检查点
             *  3.3 缓存和检查点区别
             *  3.4 检查点存储到HDFS集群
         *  4、键值对RDD数据分区
     *  二、累加器
     *  三、广播变量

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

### 一、RDD 编程 ###

#### 1、RDD序列化 ####

初始化工作是在Driver端进行的，而实际运行程序是在Executor端进行的，这就涉及到了跨进程通信，是需要序列化的。

![在这里插入图片描述][0a091392d49044acb9ab5612e1bd3800.png]

class User extends Serializable {
        
      var name: String = _
    }
    
    class Test04 {
        
      Logger.getLogger("org").setLevel(Level.ERROR)
    
      @Test
      def test(): Unit = {
        
        val conf: SparkConf = new SparkConf().setAppName("SparkCore").setMaster("local[*]")
        val sc: SparkContext = new SparkContext(conf)
        val rdd01: RDD[(Int, String)] = sc.makeRDD(Array((111, "aaa"), (222, "bbbb"), (333, "ccccc")), 3)
    
    
        val user01: User = new User()
        user01.name = "list"
        val user02: User = new User()
        user02.name = "lisi"
        val userRdd01: RDD[User] = sc.makeRDD(List(user01, user02))
    
        //  没有序列化(java.io.NotSerializableException: day04.User)
        userRdd01.foreach(user => println(user.name))
        sc.stop()
    
      }
    }

##### 1.2 Kryo序列化框架 #####

参考地址: https://github.com/EsotericSoftware/kryo

Java的序列化能够序列化任何的类。但是比较重，序列化后对象的体积也比较大。

Spark出于性能的考虑，Spark2.0开始支持另外一种Kryo序列化机制。Kryo速度是Serializable的10倍。当RDD在Shuffle数据的时候，简单数据类型、数组和字符串类型已经在Spark内部使用Kryo来序列化。

import org.apache.log4j.{
        Level, Logger}
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{
        SparkConf, SparkContext}
    import org.junit.Test
    
    class Test04 {
        
      Logger.getLogger("org").setLevel(Level.ERROR)
    
      @Test
      def test(): Unit = {
        
        val conf: SparkConf = new SparkConf().setAppName("SparkCore").setMaster("local[*]")
          // 替换默认的序列化机制
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          // 注册需要使用kryo序列化的自定义类
          .registerKryoClasses(Array(classOf[Search]))
        val sc: SparkContext = new SparkContext(conf)
    
        val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello", "world"))
        val search: Search = new Search("hello")
        val result: RDD[String] = rdd.filter(search.isMatch)
        println(result.collect().toList)
      }
    
    
    }
    
    //  关键字封装在一个类里面
    //  需要自己先让类实现序列化  之后才能替换使用kryo序列化
    class Search(val query: String) extends Serializable {
        
      def isMatch(s: String): Boolean = {
        
        s.contains(query)
      }
    }

#### 2、RDD依赖关系 ####

##### 2.1 查看血缘关系 #####

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

![在这里插入图片描述][82e658053b0e45568775971f8413fb67.png]

*  圆括号中的数字表示RDD的并行度，也就是有几个分区
 *

rdd03.toDebugString

@Test
    def test(): Unit = {
        
      val conf: SparkConf = new SparkConf().setAppName("SparkCore").setMaster("local[*]")
      val sc: SparkContext = new SparkContext(conf)
      val rdd01: RDD[String] = sc.textFile("input/1.txt")
      println(rdd01.toDebugString)
      println("rdd01===")
      val rdd02: RDD[String] = rdd01.flatMap(_.split(" "))
      println(rdd02.toDebugString)
      println("rdd02====")
      val rdd03: RDD[(String, Int)] = rdd02.map((_, 1))
      println(rdd03.toDebugString)
      println("rdd03====")
      val rdd04: RDD[(String, Int)] = rdd03.reduceByKey(_ + _)
      println(rdd04.toDebugString)
      sc.stop()
    }

![在这里插入图片描述][a0dd991da22f41daac07b8660b6826f6.png]

##### 2.2 查看依赖关系 #####

![在这里插入图片描述][56ccfb6334f547fe83b6ad3ae81dcd60.png]

![在这里插入图片描述][d46cca337be3445dbb6d48948d3148d7.png]

RDD之间的关系可以从两个维度来理解：一个是RDD是从哪些RDD转换而来，也就是 RDD的parent RDD(s)是什么（**血缘**）; 另一个就是RDD依赖于parent RDD(s)的哪些Partition(s)，这种关系就是RDD之间的依赖（**依赖**）。

RDD和它依赖的父RDD（s）的依赖关系有两种不同的类型，即窄依赖（NarrowDependency）和宽依赖（ShuffleDependency）。

##### 2.3 窄依赖 #####

一对一、多对一

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

![在这里插入图片描述][b7d6dc629f3f46619cdb6d0c0509d204.png]

##### 2.4 宽依赖 #####

一对多，会引起Shuffle

*  宽依赖表示同一个父RDD的Partition被多个子RDD的Partition依赖（只能是一对多），会引起Shuffle。
 *  总结：宽依赖我们形象的比喻为超生。
 *  具有宽依赖的*transformations*包括：*sort*、*reduceByKey*、*groupByKey*、*join*和调用*rePartition*函数的任何操作。
 *  宽依赖对Spark去评估一个transformations有更加重要的影响，比如对性能的影响。
 *  在不影响业务要求的情况下，要尽量避免使用有宽依赖的转换算子，因为有宽依赖，就一定会走shuffle，影响性能。

![在这里插入图片描述][836734da1efe4c9a858fac454bad38f1.png]

##### 2.5 Stage任务划分 #####

**DAG有向无环图**

DAG（Directed Acyclic Graph）有向无环图是由点和线组成的拓扑图形，该图形具有方向，不会闭环。

DAG记录了RDD的转换过程和任务的阶段。

**RDD任务切分**

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

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

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

@Test
    def Test(): Unit = {
        
      val conf: SparkConf = new SparkConf().setAppName("SparkCore").setMaster("local[*]")
      //  1、Application：初始化一个SparkContext即生成一个Application
      val sc: SparkContext = new SparkContext(conf)
      val lineRdd: RDD[String] = sc.textFile("input/1.txt")
      val rdd01: RDD[String] = lineRdd.flatMap(_.split(" "))
      val rdd02: RDD[(String, Int)] = rdd01.map((_, 1))
      //  3、Stage：reduceByKey算子会有宽依赖,stage阶段+1。一共2个stage
      val resultRdd: RDD[(String, Int)] = rdd02.reduceByKey(_ + _)
      //  2、Job：一个Action算子就会生成一个Job。一共2个Job
      resultRdd.collect().foreach(println)
      resultRdd.saveAsTextFile("output")
      Thread.sleep(Long.MaxValue)
      sc.stop()
    }

Application个数：

//  1、Application：初始化一个SparkContext即生成一个Application
    val sc: SparkContext = new SparkContext(conf)

Job个数

![在这里插入图片描述][96d2c90162904fb5b591ea9c9ade31bd.png]

Stage个数

![在这里插入图片描述][d26f1ae08ef04d248c5f95d5be23de8a.png]

Task数量

*  如果存在shuffle过程，系统会自动进行缓存，UI界面显示skipped的部分。
 *  从Stage中看有2个Task。

![在这里插入图片描述][984b0fb10e954c0b81e167ea11855dba.png]

![在这里插入图片描述][52cab21c55b7470d9b90861577f38f60.png]

#### 3、RDD 持久化 ####

##### 3.1 Cache缓存 #####

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

//  cache底层调用的就是persist方法,缓存级别默认用的是MEMORY_ONLY
    wortToOneRdd.cache()
    //  可以更改缓存级别
    wortToOneRdd.persist(StorageLevel.MEMORY_AND_DISK_2)

案例：

val wordRdd: RDD[String] = lineRdd.flatMap(line => line.split(" "))
    val wortToOneRdd: RDD[(String, Int)] = wordRdd.map(word => (word, 1))
    //  打印血缘关系（缓存前）
    println(wortToOneRdd.toDebugString)
    //  数据缓存
    //  cache底层调用的就是persist方法,缓存级别默认用的是MEMORY_ONLY
    wortToOneRdd.cache()
    //  可以更改缓存级别
    //    wortToOneRdd.persist(StorageLevel.MEMORY_AND_DISK_2)
    wortToOneRdd.collect().foreach(println)
    //  打印血缘关系（缓存后）
    println(wortToOneRdd.toDebugString)

![在这里插入图片描述][da51eda70f7e4018a9edcc93d7903718.png]

**缓存枚举参数**

默认的存储级别都是仅在内存存储一份。在存储级别的末尾加上“\_2”表示持久化的数据存为两份。

SER：表示序列化。

![在这里插入图片描述][b697e1d3b6704218b6d612c641c2a231.png]

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

**自带缓存`采用reduceByKey`**

// 采用reduceByKey，自带缓存
    val wordByKeyRDD: RDD[(String, Int)] = wordToOneRdd.reduceByKey(_+_)

##### 3.2 CheckPoint检查点 #####

检查点：是通过将RDD中间结果写入磁盘

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

**检查点存储路径：** Checkpoint的数据通常是存储在HDFS等容错、高可用的文件系统。

**存储格式为：** 二进制的文件。

**检查点切断血缘：** 在Checkpoint的过程中，该RDD的所有依赖于父RDD中的信息将全部被移除。

**检查点触发时间：** 对RDD进行Checkpoint操作并不会马上被执行，必须执行Action操作才能触发。但是检查点为了数据安全，会从血缘关系的最开始执行一遍。

![在这里插入图片描述][35c681bcb4084b858b791a7d3af1739e.png]

//	设置检查点数据存储路径：
    sc.setCheckpointDir("./checkpoint1")
    //	调用检查点方法：
    wordToOneRdd.checkpoint()

代码：

val rdd: RDD[String] = sc.textFile("input/1.txt")
    //  业务逻辑
    val rdd01: RDD[String] = rdd.flatMap(line => line.split(" "))
    val rdd02: RDD[(String, Long)] = rdd01.map(word => (word, System.currentTimeMillis()))
    //  增加缓存，避免再重新跑一个job做checkpoint
    rdd02.cache()
    //  数据检查点：针对wordToOneRdd做检查点计算
    rdd02.checkpoint()
    //  会立即启动一个新的job来专门的做checkpoint运算(一共会有2个job)
    rdd02.collect().foreach(println)
    //  再次触发2次执行逻辑，用来对比
    rdd02.collect().foreach(println)
    rdd02.collect().foreach(println)

执行结果：

通过页面`http://localhost:4040/jobs`查看DAG图。可以看到检查点切断了血缘依赖关系。

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>只增加checkpoint，没有增加Cache缓存打印</td> 
   <td>第1个job执行完，触发了checkpoint，第2个job运行checkpoint，并把数据存储在检查点上。第3、4个job，数据从检查点上直接读取。</td> 
  </tr> 
  <tr> 
   <td>增加checkpoint，也增加Cache缓存打印</td> 
   <td>第1个job执行完，数据就保存到Cache里面了，第2个job运行checkpoint，直接读取Cache里面的数据，并把数据存储在检查点上。第3、4个job，数据从检查点上直接读取。</td> 
  </tr> 
 </tbody> 
</table>

![在这里插入图片描述][d663dee7f9374d68b2319dc6e498cb45.png]

![在这里插入图片描述][9fe0da6df40e4a07b2a00010e5c6fe3a.png]

##### 3.3 缓存和检查点区别 #####

*  Cache缓存只是将数据保存起来，不切断血缘依赖。Checkpoint检查点切断血缘依赖。
 *  Cache缓存的数据通常存储在磁盘、内存等地方，可靠性低。Checkpoint的数据通常存储在HDFS等容错、高可用的文件系统，可靠性高。
 *  建议对checkpoint()的RDD使用Cache缓存，这样checkpoint的job只需从Cache缓存中读取数据即可，否则需要再从头计算一次RDD。
 *  如果使用完了缓存，可以通过unpersist()方法释放缓存。

##### 3.4 检查点存储到HDFS集群 #####

如果检查点数据存储到HDFS集群，要注意配置访问集群的用户名。否则会报访问权限异常。

// 设置访问HDFS集群的用户名
    System.setProperty("HADOOP_USER_NAME", "atguigu")
    
    // 需要设置路径.需要提前在HDFS集群上创建/checkpoint路径
    sc.setCheckpointDir("hdfs://hadoop102:8020/checkpoint")
    
    //  数据检查点：针对wordToOneRdd做检查点计算
    rdd02.checkpoint()

#### 4、键值对RDD数据分区 ####

Spark目前支持Hash分区、Range分区和用户自定义分区。Hash分区为当前的默认分区。分区器直接决定了RDD中分区的个数、RDD中每条数据经过Shuffle后进入哪个分区和Reduce的个数。

*  只有Key-Value类型的RDD才有分区器，非Key-Value类型的RDD分区的值是None
 *  每个RDD的分区ID范围：0~numPartitions-1，决定这个值是属于那个分区的。

val conf: SparkConf = new SparkConf().setAppName("SparkCore").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)
    //	数据源处理
    val rdd: RDD[(Int, Int)] = sc.makeRDD(List((1, 1), (2, 2), (3, 3)))
    //	打印分区器
    println(rdd.partitioner)
    //	使用HashPartitioner对RDD进行重新分区
    val rdd02: RDD[(Int, Int)] = rdd.partitionBy(new HashPartitioner(2))
    //	打印分区器
    println(rdd02.partitioner)
    sc.stop()

**Hash分区**

HashPartitioner分区的原理：对于给定的key，计算其hashCode，并除以分区的个数取余，如果余数小于0，则用余数+分区的个数（否则加0），最后返回的值就是这个key所属的分区ID。

HashPartitioner分区弊端：可能导致每个分区中数据量的不均匀，极端情况下会导致某些分区拥有RDD的全部数据。

![在这里插入图片描述][d9ab2cdb74dc494ba3b8c374a8fbe431.png]

**Ranger分区**

RangePartitioner作用：将一定范围内的数映射到某一个分区内，尽量保证每个分区中数据量均匀，而且分区与分区之间是有序的，一个分区中的元素肯定都是比另一个分区内的元素小或者大，但是分区内的元素是不能保证顺序的。简单的说就是将一定范围内的数映射到某一个分区内。

实现过程为：

第一步：先从整个RDD中采用水塘抽样算法，抽取出样本数据，将样本数据排序，计算出每个分区的最大key值，形成一个Array\[KEY\]类型的数组变量rangeBounds；

第二步：判断key在rangeBounds中所处的范围，给出该key值在下一个RDD中的分区id下标；该分区器要求RDD中的KEY类型必须是可以排序的

*  1）我们假设有100万条数据要分4个区
 *  2）从100万条中抽100个数（1,2,3, …… 100）
 *  3）对100个数进行排序，然后均匀的分为4段
 *  4）获取100万条数据，每个值与4个分区的范围比较，放入合适分区

### 二、累加器 ###

分布式共享只写变量（Executor和Executor之间不能读数据）

*  累加器用来把Executor端变量信息聚合到Driver端。在Driver中定义的一个变量，在Executor端的每个task都会得到这个变量的一份新的副本，每个task更新这些副本的值后，传回Driver端进行合并计算。
 *  注意：Executor端的任务不能读取累加器的值（例如：在Executor端调用sum.value，获取的值不是累加器最终的值）。因此我们说，累加器是一个分布式共享只写变量。

![在这里插入图片描述][c3f9aadad0af411f83a82fa21138a299.png]

//	累加器定义（SparkContext.accumulator(initialValue)方法）
    val sum: LongAccumulator = sc.longAccumulator("sum")
    //	累加器添加数据（累加器.add方法）
    sum.add(count)
    //	累加器获取数据（累加器.value）
    sum.value

val dataRdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("a", 4)))
    //  设置新的累加器
    val accSum: LongAccumulator = sc.longAccumulator("sum")
    dataRdd.foreach(line => {
        
      //  使用累加器累加
      accSum.add(line._2)
    })
    //  获取累加器，累加后的值
    println(accSum.value)

**累加器要放在行动算子中**

*  因为转换算子执行的次数取决于job的数量，如果一个spark应用有多个行动算子，那么转换算子中的累加器可能会发生不止一次更新，导致结果错误。
 *  所以，如果想要一个无论在失败还是重复计算时都绝对可靠的累加器，我们必须把它放在foreach()这样的行动算子中。
 *  对于在行动算子中使用的累加器，Spark只会把每个Job对各累加器的修改应用一次。

val value: RDD[Unit] = dataRdd.map {
        
      case (a, count) => {
        
        accSum.add(count)
      }
    }
    //假如放在map中，调用两次行动算子，map执行两次，导致最终累加器的值翻倍
    mapRDD.collect()
    mapRDD.collect()

### 三、广播变量 ###

分布式共享只读变量

广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值，以供一个或多个Spark Task操作使用。

**步骤：**

1.  调用SparkContext.broadcast（广播变量）创建出一个广播对象，任何可序列化的类型都可以这么实现。
2.  通过广播变量.value，访问该对象的值。
3.  广播变量只会被发到各个节点一次，作为只读值处理（修改这个值不会影响到别的节点）。

//	声明广播变量
    val bdStr: Broadcast[Int] = sc.broadcast(num)
    //	使用广播变量
    bdstr.value

val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6), 3)
    //  需要广播的值
    val num: Int = 1
    //  声明广播变量
    val bdStr: Broadcast[Int] = sc.broadcast(num)
    //  使用广播变量
    val rdd02: RDD[Int] = rdd.filter(lin => {
        
      lin.equals(bdStr.value)
    })
    rdd02.foreach(println)

[Spark]: https://blog.csdn.net/lydms/article/details/133974836
[Spark-Core]: https://blog.csdn.net/lydms/article/details/132653525
[Spark-Core 1]: https://blog.csdn.net/lydms/article/details/134016399
[0a091392d49044acb9ab5612e1bd3800.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/f14df16f8ddd4959a0a9c90d59bfd058.png
[82e658053b0e45568775971f8413fb67.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/5982f7897831484785630ccaed5738f0.png
[a0dd991da22f41daac07b8660b6826f6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/fa8f1e2eb1fc4f49b7ec6bf5334f7a91.png
[56ccfb6334f547fe83b6ad3ae81dcd60.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/4f80eecd46e0423084713a5169037bae.png
[d46cca337be3445dbb6d48948d3148d7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/9207258713b742a8aabdc56becb22817.png
[b7d6dc629f3f46619cdb6d0c0509d204.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/15d91d480679400f818620e83440b04e.png
[836734da1efe4c9a858fac454bad38f1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/389e33a73aab44e1a5f39f068b66e53a.png
[96d2c90162904fb5b591ea9c9ade31bd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/6a343bcd8a314c80a3df599aed5ee1b2.png
[d26f1ae08ef04d248c5f95d5be23de8a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/815d1d6193a24f3ebc53f4d85aa0e30c.png
[984b0fb10e954c0b81e167ea11855dba.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/8fb33715738c43148c03311cadf765a6.png
[52cab21c55b7470d9b90861577f38f60.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/4a8f2fa399914aeaace9bc5b693edcd5.png
[da51eda70f7e4018a9edcc93d7903718.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/e57c3b35525e4b5ab2936c29f89c8faa.png
[b697e1d3b6704218b6d612c641c2a231.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/6d9d823ddf9b4e699c1eede73d0ad4fc.png
[35c681bcb4084b858b791a7d3af1739e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/39b181546ca9440f9408138fe529f349.png
[d663dee7f9374d68b2319dc6e498cb45.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/5658ffa03acb4b058c5d6dbcc4a91b36.png
[9fe0da6df40e4a07b2a00010e5c6fe3a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/710a36493a48435f8dd396ddf8532529.png
[d9ab2cdb74dc494ba3b8c374a8fbe431.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/42ef3b7357ae4a6e82db13b867b42c1e.png
[c3f9aadad0af411f83a82fa21138a299.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/524ae7f9b4384482a9a60a82b5eaf802.png