Spark中自定义累加器_大数据培训

自定义累加器

        自定义累加器类型的功能在1.X版本中就已经提供了，但是使用起来比较麻烦，在2.0版本后，累加器的易用性有了较大的改进，而且官方还提供了一个新的抽象类：AccumulatorV2来提供更加友好的自定义类型累加器的实现方式。实现自定义类型累加器需要继承AccumulatorV2并覆写要求的方法。

       1、需求：实现一个自定义的数值累加器

       2、代码实现

        3、调用自定义累加器

package com.atguigu

import org.apache.spark.rdd.RDD

import org.apache.spark.{Accumulator, SparkConf, SparkContext}

object AccuTest {

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

    //创建SparkConf

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("AccuTest")

    //创建SC

    val sc = new SparkContext(sparkConf)

    //创建自定义累加器对象

    val accu = new MyAccu

    //注册累加器

    sc.register(accu)

    //创建RDD

    val value: RDD[Int] = sc.parallelize(Array(1, 2, 3, 4))

    //在行动算子中对累加器的值进行修改

    value.foreach { x =>

      accu.add(1)

      println(x)

    }

    //打印累加器的值

    println(accu.value)

    //关闭SparkContext

    sc.stop()

  }

}