Spark之累加器

我会带着你远行 2022-12-31 13:20 140阅读 0赞

## Spark之累加器 ##

> 累加器：分布式共享只写变量。（Executor和Executor之间不能读数据）  
> 累加器用来把Executor端变量信息聚合到Driver端。在Driver程序中定义的变量，在Executor端的每个task都会得到这个变量的一份新的副本，每个task更新这些副本的值后，传回Driver端进行merge。

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

## 1. 系统累加器 ##

**1）累加器使用**

（1）累加器定义（SparkContext.accumulator(initialValue)方法）

val sum: LongAccumulator = sc.longAccumulator("sum")

（2）累加器添加数据（累加器.add方法）

sum.add(count)

（3）累加器获取数据（累加器.value）

sum.value

**2）代码实现**

object accumulator01_system { 
    
        def main(args: Array[String]): Unit = { 
    
            //1.创建SparkConf并设置App名称
            val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")
    
            //2.创建SparkContext，该对象是提交Spark App的入口
            val sc: SparkContext = new SparkContext(conf)
    
            //3.创建RDD
            val dataRDD: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("a", 4)))
    
            //3.1 打印单词出现的次数（a,10） 代码执行了shuffle，效率比较低
            dataRDD.reduceByKey(_ + _).collect().foreach(println)
    
            //3.2 如果不用shuffle，怎么处理呢？
            var sum = 0
            // 打印是在Executor端
            dataRDD.foreach { 
                case (a, count) => { 
                    sum = sum + count
                    println("sum=" + sum)
                }
            }
            // 打印是在Driver端
            println(("a", sum))
    
            //3.3 使用累加器实现数据的聚合功能
            // Spark自带常用的累加器
            //3.3.1 声明累加器
            val sum1: LongAccumulator = sc.longAccumulator("sum1")
    
            dataRDD.foreach{ 
                case (a, count)=>{ 
                    //3.3.2 使用累加器
                    sum1.add(count)
    
                    //3.3.3 不在Executor端读取累加器的值
                    //println(sum1.value)
                }
            }
            //3.3.4 获取累加器
            println(sum1.value)
    
            //4.关闭连接
            sc.stop()
        }
    }

注意：Executor端的任务不能读取累加器的值（例如：在Executor端调用sum.value，获取的值不是累加器最终的值）。从这些任务的角度来看，累加器是一个只写变量。

**3）累加器放在行动算子中**

> 对于要在行动操作中使用的累加器，Spark只会把每个任务对各累加器的修改应用一次。因此，如果想要一个无论在失败还是重复计算时都绝对可靠的累加器，我们必须把它放在foreach()这样的行动操作中。转化操作中累加器可能会发生不止一次更新。

object accumulator02_updateCount { 
    
        def main(args: Array[String]): Unit = { 
    
            //1.创建SparkConf并设置App名称
            val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    
            //2.创建SparkContext，该对象是提交Spark App的入口
            val sc: SparkContext = new SparkContext(conf)
    
            //3.创建RDD
            val dataRDD: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("a", 4)))
    
            //3.1 定义累加器
            val sum: LongAccumulator = sc.longAccumulator("sum")
    
            val value: RDD[(String, Int)] = dataRDD.map(t => { 
                //3.2 累加器添加数据
                sum.add(1)
                t
            })
    
            //3.3 调用两次行动算子，map执行两次，导致最终累加器的值翻倍
            value.foreach(println)
            value.collect()
    
            //3.4 获取累加器的值
            println("a:"+sum.value)
    
            //4.关闭连接
            sc.stop()
        }
    }

## 2. 自定义累加器 ##

> 自定义累加器类型的功能在1.X版本中就已经提供了，但是使用起来比较麻烦，在2.0版本后，累加器的易用性有了较大的改进，而且官方还提供了一个新的抽象类：AccumulatorV2来提供更加友好的自定义类型累加器的实现方式。

**1）自定义累加器步骤**

（1）继承AccumulatorV2，设定输入、输出泛型  
（2）重写方法

**2）需求：自定义累加器，统计RDD中首字母为“H”的单词以及出现的次数。**

List(“Hello”, “Hello”, “Hello”, “Hello”, “Hello”, “Spark”, “Spark”)

**3）代码实现**

package com.spark.accumulator
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.util.AccumulatorV2
    import org.apache.spark.{ SparkConf, SparkContext}
    
    import scala.collection.mutable
    
    object accumulator03_define { 
    
        def main(args: Array[String]): Unit = { 
    
            //1.创建SparkConf并设置App名称
            val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    
            //2.创建SparkContext，该对象是提交Spark App的入口
            val sc: SparkContext = new SparkContext(conf)
    
            //3. 创建RDD
            val rdd: RDD[String] = sc.makeRDD(List("Hello", "Hello", "Hello", "Hello", "Spark", "Spark"), 2)
    
            //3.1 创建累加器
            val acc: MyAccumulator = new MyAccumulator()
    
            //3.2 注册累加器
            sc.register(acc,"wordcount")
    
            //3.3 使用累加器
            rdd.foreach(
                word =>{ 
                    acc.add(word)
                }
            )
    
            //3.4 获取累加器的累加结果
            println(acc.value)
    
            //4.关闭连接
            sc.stop()
        }
    }
    
    // 声明累加器
    // 1.继承AccumulatorV2,设定输入、输出泛型
    // 2.重新方法
    class MyAccumulator extends AccumulatorV2[String, mutable.Map[String, Long]] { 
    
        // 定义输出数据集合
        var map = mutable.Map[String, Long]()
    
        // 是否为初始化状态，如果集合数据为空，即为初始化状态
        override def isZero: Boolean = map.isEmpty
    
        // 复制累加器
        override def copy(): AccumulatorV2[String, mutable.Map[String, Long]] = { 
            new MyAccumulator()
        }
    
        // 重置累加器
        override def reset(): Unit = map.clear()
    
        // 增加数据
        override def add(v: String): Unit = { 
            // 业务逻辑
            if (v.startsWith("H")) { 
                map(v) = map.getOrElse(v, 0L) + 1L
            }
        }
    
        // 合并累加器
        override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]): Unit = { 
    
            other.value.foreach{ 
                case (word, count) =>{ 
                    map(word) = map.getOrElse(word, 0L) + count
                }
            }
        }
    
        // 累加器的值，其实就是累加器的返回结果
        override def value: mutable.Map[String, Long] = map
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221120/527046dfcd0d41c891bf9d6e02b9e059.png