Spark中Accumulator的使用

accumulator簡介

accumulator是spark提供的累加器，顧名思義，該變數只能夠增加。

只有driver能獲取到accumulator的值（使用value方法），task只能對其做增加操作（使用 +=）。你也可以在為accumulator命名（不支援python），這樣就會在spark web ui中顯示，可以幫助你了解程式執行的情況。

accumulator使用

使用示例

舉個最簡單的accumulator的使用例子：

//在driver中定義

val accum = sc.accumulator(0, "example accumulator")

//在task中進行累加

sc.parallelize(1 to 10).foreach(x=> accum += 1)

//在driver中輸出

accum.value

//結果將返回10

res: 10

累加器的錯誤用法

val accum= sc.accumulator(0, "error accumulator")
val data = sc.parallelize(1 to 10)

//用accumulator統計偶數出現的次數，同時偶數返回0，奇數返回1

val newdata = data.mapelse 1
}}

看了上面的分析，大家都有這種印象了，那就是使用累加器的過程中只能使用一次action的操作才能保證結果的準確性。

事實上，還是有解決方案的，只要將任務之間的依賴關係切斷就可以了。什麼方法有這種功能呢？你們肯定都想到了，cache，persist。呼叫這個方法的時候會將之前的依賴切除，後續的累加器就不會再被之前的transfrom操作影響到了。

val accum= sc.accumulator(0, "error accumulator")
val data = sc.parallelize(1 to 10)

//**和上方相同

val newdata = data.map}

//使用cache快取資料，切斷依賴。

newdata.cache.count

//此時accum的值為5

accum.value

newdata.foreach(println)

//此時的accum依舊是5

accum.value

總結

使用accumulator時，為了保證準確性，只使用一次action操作。如果需要使用多次則使用cache或persist操作切斷依賴。