累加器概念

https://www.yuque.com/docs/share/5314de22-204e-4731-86a7-4454a5ac0274?# 《Spark之累加器的概念和使用》

自定义累加器

自定义Map的累加器,
统计List里面的累加值,总数,平均值

  1. package com.add
  3. import org.apache.spark.rdd.RDD
  4. import org.apache.spark.util.AccumulatorV2
  5. import org.apache.spark.{SparkConf, SparkContext}
  7. object MyMapAccumulator {
  8.   def main(args: Array[String]): Unit = {
  9.     val conf: SparkConf = new SparkConf().setAppName("MyAcc").setMaster("local[4]")
  10.     val sc: SparkContext = new SparkContext(conf)
  11.     val list1 = List(30, 50, 70, 60, 10, 20, 10, 30, 40, 50)
  12.     val rdd1: RDD[Int] = sc.parallelize(list1, 2)
  14.     val acc = new MyMapAccumulator
  15.     sc.register(acc)
  16.     rdd1.foreach(x => acc.add(x))
  18.     println(acc.value)  //输出: Map(sum -> 370.0, count -> 10, avg -> 37.0)
  20.     sc.stop()
  22.   }
  23. }
  25. // 将来累加器的值同时包含 sum, count, avg
  26. // (sum, count, avg)
  27. // Map("sum"-> 1000, "count"-> 10, "avg" -> 100)
  28. class MyMapAccumulator extends AccumulatorV2[Double, Map[String, Any]] {
  29.   private var map = Map[String, Any]()
  31.   override def isZero: Boolean = map.isEmpty
  33.   override def copy(): AccumulatorV2[Double, Map[String, Any]] = {
  34.     println("copy...")
  35.     val acc = new MyMapAccumulator
  36.     acc.map = map //复制到当前的map
  37.     acc
  38.   }
  40.   /** 重置一下
  41.    * 不可变集合, 直接赋值一个空的集合
  42.    *
  43.    */
  44.   override def reset(): Unit = map = {
  45.     println("reset...")
  46.     Map[String, Any]()
  47.   }
  49.   override def add(v: Double): Unit = {
  50.     // 对sum和count进行累加. avg在最后value函数进行计算
  52.     // 强转成double进行累加.
  53.     map += "sum" -> (map.getOrElse("sum", 0D).asInstanceOf[Double] + v)
  54.     map += "count" -> (map.getOrElse("count", 0L).asInstanceOf[Long] + 1L)
  55.   }
  57.   /**
  58.    * 把所有分区的map进行合并
  59.    *
  60.    * @param other
  61.    */
  62.   override def merge(other: AccumulatorV2[Double, Map[String, Any]]): Unit = {
  63.     // 合并两个map
  64.     other match {
  65.       //是MapAcc才进行合并,如果是其它类型的话就抛异常
  66.       case o: MyMapAccumulator =>
  67.         map +=
  68.           "sum" -> (map.getOrElse("sum", 0D).asInstanceOf[Double]
  69.             + o.map.getOrElse("sum", 0D).asInstanceOf[Double])
  70.         map +=
  71.           "count" -> (map.getOrElse("count", 0L).asInstanceOf[Long]
  72.             + o.map.getOrElse("count", 0L).asInstanceOf[Long])
  73.       case _ => throw new UnsupportedOperationException
  74.     }
  75.   }
  78.   override def value: Map[String, Any] = {
  79.     // 进行平均值统计
  80.     map += "avg" ->
  81.       (map.getOrElse("sum", 0D).asInstanceOf[Double] / map.getOrElse("count", 0L).asInstanceOf[Long])
  82.     map
  83.   }
  84. }

输出

  1. Map(sum -> 370.0, count -> 10, avg -> 37.0)