Spark 累加器

文章目录

• 实现原理
• 基础编程
• • 系统累加器
  • 自定义累加器

实现原理

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

基础编程

系统累加器

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)

注意：如果是转换算子调用累加器（比如map），如果没有行动算子的话，那么不会执行！累加器一般在行动算子中执行。

自定义累加器

package acc

import org.apache.log4j.{Level, Logger}
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Spark01_Acc {
  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.ERROR)

    val conf = new SparkConf().setMaster("local").setAppName(this.getClass.getName)

    val sc = new SparkContext(conf)

    val list = List("Hello", "Spark", "Hadoop", "Scala", "Hello")

    val inputRDD = sc.makeRDD(list)

    //创建累加器对象
    val wcACC = new MyAccumulator()

    //注册累加器
    sc.register(wcACC, "WordCount")

    inputRDD.foreach(
      word => {
        wcACC.add(word)
      }
    )

    //输出累加器的值

    //获取累加器累加的结果
    println(wcACC.value)

    sc.stop()


  }

  /**
   * 自定义数据累加器，继承AccumulatorV2 定义泛型
   * in : 累加器输入的数据类型 String
   * out : 累加器返回的数据类型 mutable.Map(String,Long)
   *
   */
  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 newCnt = wcMap.getorElse(word, 0L) + 1
      wcMap.update(word, newCnt)
    }

    //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
    }
  }
}