一、Spark累加器有哪些特点
Spark累加器（Accumulator）是在Spark分布式计算框架中一个很重要的概念。它是在分布式计算过程中，允许用户在多个节点上进行分布式聚合操作的一种机制。
Spark累加器有以下几个特点：
1、只有Driver程序可以对累加器进行写操作，但所有的Worker节点都可以读取累加器的值；
2、累加器的值每次在Worker节点上更新后，都会汇总到Driver程序中；
3、累加器在多个聚合操作中可以叠加使用。
二、Spark累加器不支持自定义
Spark累加器支持两种类型的累加器：数值类型和自定义类型。但无论是哪种类型，都需要在Driver程序中进行定义。所以我们可以说Spark累加器不支持自定义。
三、Spark累加器支持数值类型
Spark累加器支持数值类型，包括整型、浮点型等。
val sc = new SparkContext(...)
val counter = sc.longAccumulator("counter")
val rdd = sc.parallelize(Array(1,2,3,4,5))
rdd.foreach(x => counter.add(x))
println(counter.value)
上面的代码定义了一个Long类型的累加器，并在并行化的RDD中遍历每个元素，将它们累加到累加器counter中。最后输出累加器的值即可。
四、Spark累加器支持自定义类型
除了数值类型，Spark累加器还支持自定义类型。自定义类型需要实现序列化接口和累加器型接口。
class MyType(var name: String) {
override def toString: String = s"MyType($name)"
}
class MyAccumulator extends AccumulatorV2[MyType, mutable.Set[MyType]] {
private val _set: mutable.Set[MyType] = mutable.Set()
override def isZero: Boolean = _set.isEmpty
override def copy(): AccumulatorV2[MyType, mutable.Set[MyType]] = {
val newAcc = new MyAccumulator()
newAcc._set.addAll(_set)
newAcc
}
override def reset(): Unit = _set.clear()
override def add(v: MyType): Unit = _set += v
override def merge(other: AccumulatorV2[MyType, mutable.Set[MyType]]): Unit = {
_set ++= other.value
}
override def value: mutable.Set[MyType] = _set
}
val myAccumulator = new MyAccumulator
sc.register(myAccumulator, "myAcc")
val rdd = sc.parallelize(Seq(MyType("a"), MyType("b"), MyType("c")))
rdd.foreach(myAccumulator.add)
println(myAccumulator.value)
上面的代码定义了一个自定义类型MyType和一个自定义类型的累加器MyAccumulator，并在注册时指定了自定义的名字”myAcc”。最后遍历RDD并将元素添加到自定义累加器中，最后输出累加器的值即可。
原创文章，作者：小蓝，如若转载，请注明出处：https://www.506064.com/n/287166.html
微信扫一扫 
支付宝扫一扫 