Spark学习——累加器(Accumulator)

累加器主要用于多个节点对同一变量进行操作，可以在executor端使用driver端定义的变量；但是executor并不能读取累加器的值。

累加器的类型 1.Accumulator[Int] 2.Accumulator[Double] 3.Accumulator[Long]等 自定义累加器 1.spark1 只需要继承AccumulatorParam，并重写 addInPlace（）//累加操作 zero（）//默认值 下面是自定义String类型的累加器（scala版本），代码如下：

object StringAccumulator extends  AccumulatorParam[String]{
          
   
//简单拼接
  override def addInPlace(r1: String, r2: String): String = {
          
    r1+" "+r2 }

  override def zero(initialValue: String): String = {
          
   ""}
}

2.spark2 相比spark1，spark2版本的累加器稍显复杂，需要继承 AccumulatorV2并重写 add（）//累加操作 isZero（）//判空 copy（）//复制 reset（）//清空 merge（）//合并 value（）//获取value

下面是自定义Json类型累加器（scala版本），代码如下：

class JsonAccumulator extends AccumulatorV2[String,JSONObject]{
          
   

  private val jsonAccumulator = new JSONObject()

  override def add(v:String): Unit = {
          
   
    val keyvalue = v.split(",")
    val key = keyvalue(0)
    val value = keyvalue(1)
    jsonAccumulator.put(key,value)
  }

  override def isZero: Boolean = {
          
   
    jsonAccumulator.isEmpty
  }

  override def copy(): AccumulatorV2[String, JSONObject] = {
          
   
    val newjsonAccumulator = new JsonAccumulator()
    import collection.JavaConversions._
    for(key<-jsonAccumulator.keySet()){
          
   
      newjsonAccumulator.add(key+","+jsonAccumulator.getString(key))
    }

    newjsonAccumulator
  }

  override def reset(): Unit = {
          
   jsonAccumulator.clear()}

  override def merge(other: AccumulatorV2[String, JSONObject]): Unit = {
          
   
    val keyset =  other.value.keySet()
    import collection.JavaConversions._
    for(key<-keyset){
          
   
      jsonAccumulator.put(key,other.value.getString(key))
    }
  }

  override def value: JSONObject = {
          
   jsonAccumulator}

}

使用累加器

（1） 使用内置累加器 （2）使用自定义累加器 spark1版本：

//上面自定义的string类型累加器
sparkContext.accumulator("")(StringAccumulator)

spark2版本：

val jsonAccumulator = new JsonAccumulator
    sparkContext.register(jsonAccumulator)

完整代码（json为例）

object test_wangty {
          
   

  def main(args: Array[String]): Unit = {
          
   
    val sparkConf = new SparkConf()
    val sparkContext = new SparkContext(sparkConf)
    val hiveContext = new HiveContext(sparkContext)
    val sqlContext = new SQLContext(sparkContext)

    val sql =
      s"""
         |select
         |group_id,
         |cast(amount as string) amount
         |from
         |test.test_wangty
         |where
         |dt = 2020-05-01
         |limit 10
      """.stripMargin

    val rs = sqlContext.sql(sql).rdd
    val jsonAccumulator = new JsonAccumulator
    sparkContext.register(jsonAccumulator)
    rs.repartition(10).foreach(x=>
      jsonAccumulator.add(x.getString(1)+","+x.getString(2))
    )

    println(jsonAccumulator.value.toJSONString)

    sparkContext.stop()



  }

}

结果如下：