spark源碼系列之累加器實現機制及自定義累加器
累加器是Spark的一種變量，顧名思義該變量只能增加。有以下特點：
1，累加器只能在Driver端構建及並只能是Driver讀取結果，Task只能累加。
2，累加器不會改變Spark Lazy計算的特點。只會在Job觸發的時候進行相關累加操作。
3，現有累加器的類型。
相信有很多學習大數據的道友，在這裏我給大家說說我滴群哦，大數據海量知識分享，784789432.在此我保證，絕對大數據的幹貨，等待各位的到來，我們一同從入門到精通吧！
二，累加器的使用
Driver端初始化，並在Action之後獲取值。
val accum = sc.accumulator(0, "test Accumulator")
Executor端進行計算
accum+=1;
三，累加器的重點類
Class Accumulator extends Accumulable
主要是實現了累加器的初始化及封裝了相關的累加器操作方法。同時在類對象構建的時候向我們的Accumulators註冊了累加器。累加器的add操作的返回值類型和我們傳入的值類型可以不一樣。所以，我們一定要定義好如何累加和合並值。也即add方法
object Accumulators:
該方法在Driver端管理著我們的累加器，也包含了特定累加器的聚合操作。
trait AccumulatorParam[T] extends AccumulableParam[T, T]:
AccumulatorParam的addAccumulator操作的泛型封裝，具體的實現還是要再具體實現類裏面實現addInPlace方法。
object AccumulatorParam:
主要是進行隱式類型轉換的操作。
TaskContextImpl:
在Executor端管理著我們的累加器。
四，累加器的源碼解析
1,Driver端的初始化
val accum = sc.accumulator(0, "test Accumulator")
val acc = new Accumulator(initialValue, param, Some(name))
主要是在Accumulable(Accumulator)中調用了，這樣我們就可以使用Accumulator使用了。
Accumulators.register(this)
2,Executor端的反序列化得到我們對象的過程
首先，我們的value_ 可以看到其並不支持序列化
@volatile @transient private var value_ : R = initialValue // Current value on master
其初始化是在我們反序列化的時候做的，反序列化還完成了Accumulator向我們的TaskContextImpl的註冊
反序列化是在調用ResultTask的RunTask方法的時候做的
val (rdd, func) = ser.deserialize[(RDD[T], (TaskContext, Iterator[T]) => U)](
ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)
過程中會調用
private def readObject(in: ObjectInputStream): Unit = Utils.tryOrIOException {
in.defaultReadObject()
value_ = zero
deserialized = true
// Automatically register the accumulator when it is deserialized with the task closure.
//
// Note internal accumulators sent with task are deserialized before the TaskContext is created
// and are registered in the TaskContext constructor. Other internal accumulators, such SQL
// metrics, still need to register here.
val taskContext = TaskContext.get()
if (taskContext != null) {
taskContext.registerAccumulator(this)
}
}
3,累加器的累加
accum+=1;
param.addAccumulator(value_, term)
根據不同的累加器參數有不同的實現AccumulableParam
如，int類型。最終調用的AccumulatorParam特質的addAccumulator方法。
trait AccumulatorParam[T] extends AccumulableParam[T, T] {
def addAccumulator(t1: T, t2: T): T = {
addInPlace(t1, t2)
}
}
然後，調用的是各個具體實現的addInPlace方法
implicit object IntAccumulatorParam extends AccumulatorParam[Int] {
def addInPlace(t1: Int, t2: Int): Int = t1 + t2
def zero(initialValue: Int): Int = 0
}
返回後更新了我們的Accumulators的value_的值。
4,Accumulator的各個節點累加的之後的聚合操作
在Task類的run方法裏面得到並返回的
(runTask(context), context.collectAccumulators())
最終在DAGScheduler裏面調用了updateAccumulators(event)
在updateAccumulators方法中
Accumulators.add(event.accumUpdates)
具體內容如下:
def add(values: Map[Long, Any]): Unit = synchronized {
for ((id, value) <- values) {
if (originals.contains(id)) {
// Since we are now storing weak references, we must check whether the underlying data
// is valid.
originals(id).get match {
case Some(accum) => accum.asInstanceOf[Accumulable[Any, Any]] ++= value
case None =>
throw new IllegalAccessError("Attempted to access garbage collected Accumulator.")
}
} else {
logWarning(s"Ignoring accumulator update for unknown accumulator id $id")
}
}
}
5,最後我們就可以獲取到累加器的值了
accum.value
五，累加器使用註意事項
累加器不會改變我們RDD的Lazy的特性，之後再Action之後完成計算和更新。
但是假如出現兩個Action公用一個轉化操作，如map，在map裏面進行累加器累加，那麽每次action都會累加，造成某些我們不需要的結果。
六，自定義累加器
自定義累加器輸出
七，總結
主要牽涉點就是序列化及類加載執行，這是深入玩spark的必須.spark源碼系列之累加器實現機制及自定義累加器