Spark累加器实现原理及基础编程

青旅半醒 2023-10-03 14:22 26阅读 0赞

### Spark累加器实现原理及基础编程 ###

#### 实现原理 ####

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

#### 基础编程 ####

##### 系统累加器：longAccumulator #####

> 案例：实现求1,2,3,4,5的累加之和。

package com.zxl.spark.core
    
    import org.apache.spark.{
        SparkConf, SparkContext}
    
    /*
    * 累加器
    * */
    object AccumulatorDemo {
        
      def main(args: Array[String]): Unit = {
        
        val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("AccumulatorDemo")
        val sc: SparkContext = new SparkContext(conf)
        val rdd = sc.makeRDD(List(1,2,3,4,5))
        // 声明累加器
        var sum = sc.longAccumulator("sum")
        rdd.foreach(
          num => {
        
            // 使用累加器
            sum.add(num)
          } )
        // 获取累加器的值
        println("sum = " + sum.value)
      }
    }

###### 运行结果 ######

sum = 15

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

> 案例：自定义累加器实现wordcount

package com.zxl.spark.core
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.util.AccumulatorV2
    import org.apache.spark.{
        SparkConf, SparkContext}
    
    import scala.collection.mutable
    
    /**
     * 自定义累加器
     * */
    object AccumulatorDemo2 {
        
      def main(args: Array[String]): Unit = {
        
        val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("AccumulatorDemo2")
        val sc: SparkContext = new SparkContext(conf)
        val rdd: RDD[String] = sc.makeRDD(List("hello", "zxl", "spark", "hello"))
        val wordCountAccumulator: WordCountAccumulator = new WordCountAccumulator()
        sc.register(wordCountAccumulator,"wcAcc")
        rdd.foreach(
          word=>{
        
            wordCountAccumulator.add(word)
          }
        )
        println(wordCountAccumulator.value)
      }
    
    }
    
    // 自定义累加器
    // 1. 继承 AccumulatorV2，并设定泛型
    // 2. 重写累加器的抽象方法
    class WordCountAccumulator extends AccumulatorV2[String, mutable.Map[String,Long]]{
        
      var map : mutable.Map[String, Long] = mutable.Map()
      // 累加器是否为初始状态
      override def isZero: Boolean = {
        
        map.isEmpty
      }
      // 复制累加器
      override def copy(): AccumulatorV2[String, mutable.Map[String, Long]] = {
        
        new WordCountAccumulator
      }
      // 重置累加器
      override def reset(): Unit = {
        
        map.clear()
      }
      // 向累加器中增加数据 (In)
      override def add(word: String): Unit = {
        
        // 查询 map 中是否存在相同的单词
        // 如果有相同的单词，那么单词的数量加 1
        // 如果没有相同的单词，那么在 map 中增加这个单词
        map(word) = map.getOrElse(word, 0L) + 1L
      }
      // 合并累加器
      override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]):
      Unit = {
        
        val map1 = map
        val map2 = other.value
        // 两个 Map 的合并
        map = map1.foldLeft(map2)(
          ( innerMap, kv ) => {
        
            innerMap(kv._1) = innerMap.getOrElse(kv._1, 0L) + kv._2
            innerMap
          }
        ) }
      // 返回累加器的结果 （Out）
      override def value: mutable.Map[String, Long] = map
    }

###### 运行结果 ######

Map(spark -> 1, zxl -> 1, hello -> 2)