Spark-核心编程（六）累加器与广播变量

╰+攻爆jí腚メ 2022-09-03 04:25 173阅读 0赞

## 累加器 ##

分布式共享只写变量，Executor之间是不能互相访问到累加器的，Driver可以获取到Executor中累加器的数据。

### 实现原理 ###

累加器用来把 Executor 端变量信息聚合到 Driver 端。在 Driver 程序中定义的变量,在Executor 端的每个 Task 都会得到这个变量的一份新的副本,每个 task 更新这些副本的值后,传回 Driver 端进行 merge。

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

### 基础编程 ###

val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
    
            // reduce包括了分区内计算和分区间计算
            println(rdd.reduce(_ + _)) // 10
    
            var sum = 0
            rdd.foreach(sum += _)
            println(sum)               // 0

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

#### 系统累加器 ####

// 获取系统累加器
    // Spark默认就提供了简单数据聚合的累加器
    val sumAcc: LongAccumulator = sc.longAccumulator("sum")
    
    // sc.doubleAccumulator
    // sc.collectionAccumulator
    
    // 使用累加器
    rdd.foreach(sumAcc.add(_))
    
    // 获取累加器的值
    println(sumAcc.value)   // 10
    
    // 少加：转换算子中调用累加器，如果没有行动算子的话，那么不会执行
    val sumAcc1: LongAccumulator = sc.longAccumulator("sum1")
    rdd.map(sumAcc1.add(_))
    println(sumAcc1.value)   // 0
    
    // 多加
    val sumAcc2: LongAccumulator = sc.longAccumulator("sum1")
    rdd.map(sumAcc2.add(_)).collect()
    rdd.map(sumAcc2.add(_)).collect()
    println(sumAcc2.value)   // 20
    
    // 一般情况下，累加器，会放到行动算子中进行操作

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

#### 自定义累加器 ####

object TestACCWorkCount { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
            val sc = new SparkContext(sparkConf)
    
            val rdd = sc.makeRDD(List("Hello", "word", "Hello", "Spark"))
    
            // 基本的wordCount
    // rdd.map((_, 1)).reduceByKey(_ + _)
    
            // 累加器: WordCount
            // 创建累加器对象，向Spark进行注册
            var wcAcc = new MyAccumulator()
            sc.register(wcAcc, "wordCountAcc")
    
            // 使用累加器
            rdd.foreach(wcAcc.add(_))
    
            // 获取累加器累加的记过
            println(wcAcc.value)
            // Map(Hello -> 2, Spark -> 1, word -> 1)
    
            sc.stop()
        }
    
        /** * 自定义累计器 * 1、继承AccumulatorV2，定义泛型：IN，累加器输入数据类型，OUT，累加器输出类型 */
        class MyAccumulator extends AccumulatorV2[String, mutable.Map[String, Long]] { 
    
            private var wcMap = mutable.Map[String, Long]()
    
            // 判断是否出事状态
            override def isZero: Boolean = wcMap.isEmpty
    
            override def copy(): AccumulatorV2[String, mutable.Map[String, Long]] = new MyAccumulator
    
            // 重置
            override def reset(): Unit = wcMap.clear()
    
            // 获取累加器需要计算的值
            override def add(word: String): Unit = { 
    
                val newCount: Long = wcMap.getOrElse(word, 0L) + 1L
                wcMap.update(word, newCount)
            }
    
            // Driver合并多个累加器
            override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]): Unit = { 
    
                val map1 = this.wcMap
                val map2 = other.value
    
                map2.foreach { 
                    case (word, count) => { 
                        val newCount = map1.getOrElse(word, 0L) + count
                        map1.update(word, newCount)
                    }
                }
            }
    
            // 获取累加器的结果
            override def value: mutable.Map[String, Long] = wcMap
        }
    }

## 广播变量 ##

### 实现原理 ###

广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值,以供一个或多个 Spark 操作使用。比如,如果你的应用需要向所有节点发送一个较大的只读查询表,广播变量用起来都很顺手。在多个并行操作中使用同一个变量,但是 Spark 会为每个任务分别发送。

闭包数据，都是以Task为单位发送的，每个任务中包含闭包数据，这样可能会导致，一个Executor中含有大量重复的数据，并且占用大量的内存

Executor其实就是一个JVM，所以在启动时，会自动分配内存，完全可以将任务中的闭包数据放置在Executor的内存中，达到共享的目的

Spark中的官博变量就可以将闭包的数据保存到Executor的内存中，Spark中的广播变量不能够更改，因为广播变量是分布式只读变量

object TestBc { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
            val sc = new SparkContext(sparkConf)
    
            var rdd1 = sc.makeRDD(List(
                ("a", 1),
                ("b", 2),
                ("c", 3),
            ))
    
            val rdd2 = sc.makeRDD(List(
                ("a", 4),
                ("b", 5),
                ("c", 6),
            ))
    
            // join会导致数据量集合增长，并且会影响shuffle的性能，不推荐使用
            rdd1.join(rdd2).collect.foreach(println)
    
            println("=======")
    
            val map = mutable.Map(
                ("a", 4),
                ("b", 5),
                ("c", 6),
            )
    
            rdd1.map { 
                case (w, c) => { 
                    val newC: Int = map.getOrElse(w, 0)
                    (w, (c, newC))
                }
            }.collect.foreach(println)
    
            println("=======")
    
            // 封装广播变量
            val bc: Broadcast[mutable.Map[String, Int]] = sc.broadcast(map)
    
            rdd1.map { 
                case (w, c) => { 
                    // 访问广播变量
                    val newC: Int = bc.value.getOrElse(w, 0)
                    (w, (c, newC))
                }
            }.collect.foreach(println)
    
            sc.stop()
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]: /images/20220829/bf7cd2944fd540a6b3a6ecad6897d861.png
[525441d5cae0410ba5ca648787a2058b.png]: /images/20220829/d17c25fcbf8a497292573e623e9542e6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]: /images/20220829/cf5e2eebd23b4c39bb4e9d4a279621ac.png