1. 程式人生 > >spark源碼系列之累加器實現機制及自定義累加器

spark源碼系列之累加器實現機制及自定義累加器

大數據 spark

一,基本概念

累加器是Spark的一種變量,顧名思義該變量只能增加。有以下特點:

1,累加器只能在Driver端構建及並只能是Driver讀取結果,Task只能累加。

2,累加器不會改變Spark Lazy計算的特點。只會在Job觸發的時候進行相關累加操作。

3,現有累加器的類型。
相信有很多學習大數據的道友,在這裏我給大家說說我滴群哦,大數據海量知識分享,784789432.在此我保證,絕對大數據的幹貨,等待各位的到來,我們一同從入門到精通吧!

二,累加器的使用

Driver端初始化,並在Action之後獲取值。

val accum = sc.accumulator(0, "test Accumulator")

accum.value

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源碼系列之累加器實現機制及自定義累加器