Spark实现自定义累计器

灰太狼 2022-01-26 15:05 292阅读 0赞

## 前述 ##

Spark中因为算子中的真正逻辑是发送到Executor中去运行的，所以当Executor中需要引用外部变量时，需要使用广播变量。累机器相当于统筹大变量，常用于计数，统计。

**注意事项**  
1、能不能将一个RDD使用广播变量广播出去？  
不能，因为RDD是不存储数据的。可以将RDD的结果广播出去。  
2、 广播变量只能在Driver端定义，不能在Executor端定义。  
3、 在Driver端可以修改广播变量的值，在Executor端无法修改广播变量的值。  
4、如果executor端用到了Driver的变量，如果不使用广播变量在Executor有多少task就有多少Driver端的变量副本。  
5、如果Executor端用到了Driver的变量，如果使用广播变量在每个Executor中只有一份Driver端的变量副本。

## 自定义累加器实现代码 ##

import org.apache.spark.util.AccumulatorV2
    import scala.collection.mutable
    
    /**
      * 自定义累加器
      */
    class SessionAggrStatAccumulator extends AccumulatorV2[String, mutable.HashMap[String, Int]] {
    
      // 保存所有聚合数据
      private val aggrStatMap = mutable.HashMap[String, Int]()
    
      override def isZero: Boolean = {
        aggrStatMap.isEmpty
      }
    
      override def copy(): AccumulatorV2[String, mutable.HashMap[String, Int]] = {
        val newAcc = new SessionAggrStatAccumulator
        aggrStatMap.synchronized{
          newAcc.aggrStatMap ++= this.aggrStatMap
        }
        newAcc
      }
    
      override def reset(): Unit = {
        aggrStatMap.clear()
      }
    
      override def add(v: String): Unit = {
        if (!aggrStatMap.contains(v))
          aggrStatMap += (v -> 0)
        aggrStatMap.update(v, aggrStatMap(v) + 1)
      }
    
      override def merge(other: AccumulatorV2[String, mutable.HashMap[String, Int]]): Unit = {
        other match {
          case acc:SessionAggrStatAccumulator => {
            (this.aggrStatMap /: acc.value){ case (map, (k,v)) => map += ( k -> (v + map.getOrElse(k, 0)) )}
          }
        }
      }
    
      override def value: mutable.HashMap[String, Int] = {
        this.aggrStatMap
      }
    }

**注册自定义累计器**

// 构建Spark上下文
        val sparkConf = new SparkConf().setAppName("SessionAnalyzer").setMaster("local[*]")
       // 创建Spark客户端
        val spark = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()
        val sc = spark.sparkContext
        // 设置自定义累加器，实现所有数据的统计功能,注意累加器也是懒执行的
        val sessionAggrStatAccumulator = new SessionAggrStatAccumulator
    
        // 注册自定义累加器
        sc.register(sessionAggrStatAccumulator, "sessionAggrStatAccumulator")

至此，自定义累加器就完成了