7.4 Spark Core 闭包依赖关系序列化

朴灿烈づ我的快乐病毒、 2022-12-08 15:35 15阅读 0赞

### 文章目录 ###

*  0 案例实操
 *  一. 序列化
 *   *  二. 闭包 closure
 *  三. 序列化
 *   *   *  3.1 Kryo序列化框架
 *  三. 依赖关系
 *   *   *   *  NarrowDependency 窄依赖
             *  ShuffleDependency 宽依赖
             *  Dependency总结

# 0 案例实操 #

统计每一个省份每个广告被点击数量排行TOP

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 1]

转换一下，把省份单独列出来

（河北省，bbb）,2            ----------->           河北省,（bbb,2）

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 2]

相同省份的聚合

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 3]

总览  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 4]

//TODO 统计出每一个省份每个广告被点击数量排行的top3
      //TODO 1.获取原始数据
      val rowData = sc.textFile("D:\\workplace\\ifeng-Spark\\ifeng-Spark-core\\data\\agent.log")
    
      //TODO 2.将原始数据进行结构转换，((省份-广告)，1)
      //rowData.flatMap(_.split(" "))
      val Data1 = rowData.map(
        line => {
        
          val datas = line.split(" ")
          (datas(1) + "-" + datas(4), 1)
        }
      )
    
      //TODO 3.将相同key的数据进行分组聚合, ((省份-广告)，sum)
      val reduceData = Data1.reduceByKey(_ + _)
    
    
      //TODO 4.将聚合后的结果进行结构的转换, (省份,(广告,sum))
      val recudeRDD = reduceData.map {
        
        case (key, sum) => {
        
          val keys = key.split("-")
          (keys(0), (keys(1), sum))
        }
      }
    
        //TODO 5.相同省份的数据分在一个组中, (省份,Iterator[(广告1,sum1),(广告2,sum2)])
        val RDD1: RDD[(String, Iterable[(String, Int)])] = recudeRDD.groupByKey()
    
    
      //TODO 6.只取前三位 (省份,Iterator[(广告1,sum1),(广告2,sum2)])
      val sortRDD = RDD1.mapValues(iter => {
        
        iter.toList.sortWith(
          (left, right) => {
        
            left._2 > right._2
          }
        ).take(3)
      })
    
        //TODO 打印规范数据
    
        val result: Array[(String, List[(String, Int)])] = sortRDD.collect()
        println(result)
    
      }
    }

# 一. 序列化 #

//TODO Spark序列化
    //    val RDD = sc.parallelize(List(1, 2, 3, 4, 5))
    //    RDD.foreach(
    //      num => {
        
    //        val user = new User()
    //
    //        println("age = " + (user.age + num))
    //
    //      }
    //    )
        /*
          age = 24
          age = 21
          age = 25
          age = 22
          age = 23    
        */
    
        // TODO 把 User对象拿到循环外
    
        val RDD = sc.parallelize(List(1, 2, 3, 4, 5))
        val user = new User()
        RDD.foreach(
        num => {
        
            println("age = " + (user.age + num))
          }
        )

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 5]

foreach是RDD 的算子，在executor端执行  
而创建User对象是在Dirver端执行的，所以需要序列化 ，而Uesr不能序列化

//TODO 如果算子中使用了算子外的对象,那么执行时间,需要保证可序列化
      class User() extends Serializable {
        
        val age = 20
      }

或者直接用样例类，底成实现了Serializable

case class User(age:Int = 20)

## 二. 闭包 closure ##

及时List为空 ，依然会报未序列化

val user = new User()
        val RDD = sc.parallelize( List() )
    
      class User() {
        
        val age = 20
      }

函数用到了 函数外的变量， 函数 和 变量

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 7]

**解析**

(y:int) => x - y
    
    返回是一个匿名函数，因为该函数引用到函数外的 x , 那么该函数和x整体形成一个闭包
    例如 val f = minusxy(20) 就是一个闭包
    
    1.可以理解 返回函数就是一个对象，x就是该对象的一个字段，共同形成闭包
    2.当多次调用 f 时 (多次调用闭包)，发现使用的是同一个 x ，所以 x 不变
    3.在使用闭包时 ， 主要搞清楚返回函数引用了函数外的哪些变量，因为会形成一个整体(实体) ， 形成闭包

**闭包的优缺点**

1.  返回的匿名函数和makeSuffix(suffix string) 的 suffix 变量 组合形成一个闭包 ，因为返回的函数引用到suffix这个变量
2.  我们体会一下闭包的好处 ， 如果使用传统的方法，也可以轻松实现这个功能，但是每次都需要传入后缀，而闭包因为可以保留上次引用的某个值。

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

//Spark的算子操作其实都是闭包,所以闭包有可能包含外部的变量
    //如果包含外部的变量,那么就一定要保证这个外部变量可序列化
    //所以Spark在提交作业之前,对闭包内的变量进行检测,检测是否能够序列化
    //此操作称为 闭包检测

val user = new User()
        val RDD = sc.parallelize( List() )
        //Spark的算子操作其实都是闭包,所以闭包有可能包含外部的变量
        //如果包含外部的变量,那么就一定要保证这个外部变量可序列化
        //所以Spark在提交作业之前,对闭包内的变量进行检测,检测是否能够序列化
        //此操作称为 闭包检测 
        RDD.foreach(
          num => {
        
            println("age = " + (user.age ))
          }
        )
    
        sc.stop()
      }

# 三. 序列化 #

**闭包检测**  
从计算的角度，算子以外的代码都是在Driver端执行的，算子内的代码在executor端执行的。  
导致 算子内 经常会用到 算子外 的数据，这样就形成了闭包的效果，  
如果算子外的数据无法序列化，那么就无法传递给executor端执行，  
所以在执行任务前，需要检测闭包内的对象是否可以进行序列化

Scala2.12版本后闭包编译方式发生了改变

String已经完成了序列化  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 8]

### 3.1 Kryo序列化框架 ###

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

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

object serializable_Kryo {
        
    
        def main(args: Array[String]): Unit = {
        
    
            val conf: SparkConf = new SparkConf()
                    .setAppName("SerDemo")
                    .setMaster("local[*]")
                    // 替换默认的序列化机制
                    .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                    // 注册需要使用 kryo 序列化的自定义类
                    .registerKryoClasses(Array(classOf[Searcher]))
    
            val sc = new SparkContext(conf)
    
            val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello atguigu", "atguigu", "hahah"), 2)
    
            val searcher = new Searcher("hello")
            val result: RDD[String] = searcher.getMatchedRDD1(rdd)
    
            result.collect.foreach(println)
        }
    }
    case class Searcher(val query: String) {
        
    
        def isMatch(s: String) = {
        
            s.contains(query)
        }
    
        def getMatchedRDD1(rdd: RDD[String]) = {
        
            rdd.filter(isMatch) 
        }
    
        def getMatchedRDD2(rdd: RDD[String]) = {
        
            val q = query
            rdd.filter(_.contains(q))
        }
    }

# 三. 依赖关系 #

查看基类

/**
     * :: DeveloperApi ::
     * Base class for dependencies.
     */
    @DeveloperApi
    abstract class Dependency[T] extends Serializable {
        
      def rdd: RDD[T]
    }

底层还是RDD

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 9]

#### NarrowDependency 窄依赖 ####

一个父RDD的partition的 最多 被 子RDD使用一次

NarrowDependency 也是抽象类，三个实现类

/**
     * :: DeveloperApi ::
     * Base class for dependencies where each partition of the child RDD depends on a small number
     * of partitions of the parent RDD. Narrow dependencies allow for pipelined execution.
     */
    @DeveloperApi
    abstract class NarrowDependency[T](_rdd: RDD[T]) extends Dependency[T] {
        
      /**
       * Get the parent partitions for a child partition.
       * @param partitionId a partition of the child RDD
       * @return the partitions of the parent RDD that the child partition depends upon
       */
      def getParents(partitionId: Int): Seq[Int]
    
      override def rdd: RDD[T] = _rdd
    }

**OneToOneDenpendency**

/**
     * :: DeveloperApi ::
     * Represents a one-to-one dependency between partitions of the parent and child RDDs.
     */
    @DeveloperApi
    class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd) {
        
      override def getParents(partitionId: Int): List[Int] = List(partitionId)
    }

一对一依赖  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 10]

map filter union join都是OneToOne依赖

def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))] = self.withScope {
        
        this.cogroup(other, partitioner).flatMapValues( pair =>	// 调用cogroup
          for (v <- pair._1.iterator; w <- pair._2.iterator) yield (v, w)
        )
      }
    
      def cogroup[W](other: RDD[(K, W)], partitioner: Partitioner)
          : RDD[(K, (Iterable[V], Iterable[W]))] = self.withScope {
        
        if (partitioner.isInstanceOf[HashPartitioner] && keyClass.isArray) {
        
          throw new SparkException("HashPartitioner cannot partition array keys.")
        }
        val cg = new CoGroupedRDD[K](Seq(self, other), partitioner) // new 一个CoGroupRDD对象
        cg.mapValues {
         case Array(vs, w1s) =>
          (vs.asInstanceOf[Iterable[V]], w1s.asInstanceOf[Iterable[W]])
        }
      }

join 对两个需要连接的 RDD 进行 cogroup函数操作，将相同 key 的数据能够放到一个分区，在 cogroup 操作之后 形成的新 RDD 对每个key 下的元素进行笛卡尔积的操作，返回的结果再展平，对应 key 下的所有元组形成一个集 合。最后返回 RDD\[(K， (V， W))\];

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 11]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 12]

1. 两个父RDD 分区数 和 子 分区数 一致时间，不会产生shuffle
    2. 子 分区数 和 其中一个父 RDD分区数一致， 不一样分区数的RDD 产生shuffle
    3. 三个RDD分区数都不同，必然shuffle

val left = sc.parallelize(List(("小红","一班"),("xiaolan","二班"),("xiaogang","一班"))).partitionBy(new HashPartitioner(3))
        val right = sc.parallelize(List(("小红",54),("xiaolan",6854),("xiaogang",948))).partitionBy(new HashPartitioner(3))
    
        left.join(right).foreach(println)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 13]

val left = sc.parallelize(List(("小红","一班"),("xiaolan","二班"),("xiaogang","一班")),3).partitionBy(new HashPartitioner(3))
        val right = sc.parallelize(List(("小红",54),("xiaolan",6854),("xiaogang",948)),3).partitionBy(new HashPartitioner(3))
    
        val left2 = left.groupByKey()
        
        left2.join(right).foreach(println)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 14]

#### ShuffleDependency 宽依赖 ####

一个父RDD的分区会被子类RDD的分区使用多次

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 15]

#### Dependency总结 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 16]

**WC解析**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/34f5a41a28004d638fee6cd8902c6a8c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/4a59a11e84d24b55b7474dce20a42821.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/f2651bf93811419eb365eb97962b7ede.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/38c13bafe03f455e85c23797a110725f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/77d11456cceb46ddb7ba21a89132a1d1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/ad4af2a1721a4657b1559e0ca80a24c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/df4dfcda20a4497aabdfa9a6763ee78f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221123/1132015b37fb417898b753c20d3b17a4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221123/af6423cfc63440b496b6c12e1bcdfb93.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221123/1d037d4a72be46a89726ded4d95f9a0e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20221123/e2958a97754b45e398cfbcac64f12027.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 11]: /images/20221123/14f1d11a437143cca7e13151f452b43c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 12]: /images/20221123/6c983f203b54466981434feb62cb1f4c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 13]: /images/20221123/2f3d0c341fef4f33b474bdfc05f2b827.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 14]: /images/20221123/ad56cf2629a34d3687ee5959393d1f91.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 15]: /images/20221123/911139a43f094f8aa74774ba3dfe1b9a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTM4MTgzMw_size_16_color_FFFFFF_t_70_pic_center 16]: /images/20221123/6cb4ffc8a4da46babf7063ccf75507b6.png