生产常用Spark累加器剖析之三(自定义累加器)

思路 & 需求

参考IntAccumulatorParam的实现思路（上述文章中有讲）：

trait AccumulatorParam[T] extends AccumulableParam[T, T] {  def addAccumulator(t1: T, t2: T): T = {    // addInPlace有很多具体的实现类    // 如果想要实现自定义的话，就得实现这个方法    addInPlace(t1, t2)  }}

自定义也可以通过这个方法去实现，从而兼容我们自定义的累加器

需求：这里实现一个简单的案例，用分布式的方法去实现随机数

**  * 自定义的AccumulatorParam  *  * Created by lemon on 2018/7/28.  */object UniqueKeyAccumulator extends AccumulatorParam[Map[Int, Int]] {  override def addInPlace(r1: Map[Int, Int], r2: Map[Int, Int]): Map[Int, Int] = {      // ++用于两个集合相加      r1++r2    }    override def zero(initialValue: Map[Int, Int]): Map[Int, Int] = {      var data: Map[Int, Int] = Map()      data    }}/**  * 使用自定义的累加器，实现随机数  *  * Created by lemon on 2018/7/28.  */object CustomAccumulator {  def main(args: Array[String]): Unit = {    val sparkConf = new SparkConf().setAppName("CustomAccumulator").setMaster("local[2]")    val sc = new SparkContext(sparkConf)    val uniqueKeyAccumulator = sc.accumulable(Map[Int, Int]())(UniqueKeyAccumulator)    val distData = sc.parallelize(1 to 10)    val mapCount = distData.map(x => {      val randomNum = new Random().nextInt(20)      // 构造一个k-v对      val map: Map[Int, Int] = Map[Int, Int](randomNum -> randomNum)      uniqueKeyAccumulator += map    })    println(mapCount.count())    // 获取到累加器的值 中的key值，并进行打印    uniqueKeyAccumulator.value.keys.foreach(println)    sc.stop()  }}

运行结果如下图：## 思路 & 需求

参考IntAccumulatorParam的实现思路（上述文章中有讲）：

trait AccumulatorParam[T] extends AccumulableParam[T, T] {  def addAccumulator(t1: T, t2: T): T = {    // addInPlace有很多具体的实现类    // 如果想要实现自定义的话，就得实现这个方法    addInPlace(t1, t2)  }}

自定义也可以通过这个方法去实现，从而兼容我们自定义的累加器

需求：这里实现一个简单的案例，用分布式的方法去实现随机数

**  * 自定义的AccumulatorParam  *  * Created by lemon on 2018/7/28.  */object UniqueKeyAccumulator extends AccumulatorParam[Map[Int, Int]] {  override def addInPlace(r1: Map[Int, Int], r2: Map[Int, Int]): Map[Int, Int] = {      // ++用于两个集合相加      r1++r2    }    override def zero(initialValue: Map[Int, Int]): Map[Int, Int] = {      var data: Map[Int, Int] = Map()      data    }}/**  * 使用自定义的累加器，实现随机数  *  * Created by lemon on 2018/7/28.  */object CustomAccumulator {  def main(args: Array[String]): Unit = {    val sparkConf = new SparkConf().setAppName("CustomAccumulator").setMaster("local[2]")    val sc = new SparkContext(sparkConf)    val uniqueKeyAccumulator = sc.accumulable(Map[Int, Int]())(UniqueKeyAccumulator)    val distData = sc.parallelize(1 to 10)    val mapCount = distData.map(x => {      val randomNum = new Random().nextInt(20)      // 构造一个k-v对      val map: Map[Int, Int] = Map[Int, Int](randomNum -> randomNum)      uniqueKeyAccumulator += map    })    println(mapCount.count())    // 获取到累加器的值 中的key值，并进行打印    uniqueKeyAccumulator.value.keys.foreach(println)    sc.stop()  }}

运行结果如下图：