RDD编程进阶自定义累加器_大数据培训

自定义累加器

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

• 需求：实现一个自定义的数值累加器
• 代码实现

package com.atguigu

import org.apache.spark.util.AccumulatorV2

class MyAccu extends AccumulatorV2[Int, Int] {

  var sum = 0

  //判断是否为空

  override def isZero: Boolean = sum == 0

  //复制

  override def copy(): AccumulatorV2[Int, Int] = {

    val accu = new MyAccu

    accu.sum = this.sum

    accu

  }

  //重置

  override def reset(): Unit = sum = 0

  //累加

  override def add(v: Int): Unit = sum += v

  //合并

  override def merge(other: AccumulatorV2[Int, Int]): Unit = sum += other.value

  //返回值

  override def value: Int = sum

}

• 调用自定义累加器

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()

  }

}。