Spark 累加器與廣播變數

一、簡介

在 Spark 中，提供了兩種型別的共享變數：累加器 (accumulator) 與廣播變數 (broadcast variable)：

累加器：用來對資訊進行聚合，主要用于累計計數等場景；
廣播變數：主要用于在節點間高效分發大物件，

二、累加器

這里先看一個具體的場景，對于正常的累計求和，如果在集群模式中使用下面的代碼進行計算，會發現執行結果并非預期：

var counter = 0
val data = https://www.cnblogs.com/hulichao/p/Array(1, 2, 3, 4, 5)
sc.parallelize(data).foreach(x => counter += x)
 println(counter)

counter 最后的結果是 0，導致這個問題的主要原因是閉包，

file

2.1 理解閉包

1. Scala 中閉包的概念

這里先介紹一下 Scala 中關于閉包的概念：

var more = 10
val addMore = (x: Int) => x + more

如上函式 addMore 中有兩個變數 x 和 more:

x : 是一個系結變數 (bound variable)，因為其是該函式的入參，在函式的背景關系中有明確的定義；
more : 是一個自由變數 (free variable)，因為函式字面量本生并沒有給 more 賦予任何含義，

按照定義：在創建函式時，如果需要捕獲自由變數，那么包含指向被捕獲變數的參考的函式就被稱為閉包函式，

2. Spark 中的閉包

也可以參考：https://blog.csdn.net/hu_lichao/article/details/112451982

在實際計算時，Spark 會將對 RDD 操作分解為 Task，Task 運行在 Worker Node 上，在執行之前，Spark 會對任務進行閉包，如果閉包內涉及到自由變數，則程式會進行拷貝，并將副本變數放在閉包中，之后閉包被序列化并發送給每個執行者，因此，當在 foreach 函式中參考 counter 時，它將不再是 Driver 節點上的 counter，而是閉包中的副本 counter，默認情況下，副本 counter 更新后的值不會回傳到 Driver，所以 counter 的最終值仍然為零，

需要注意的是：在 Local 模式下，有可能執行 foreach 的 Worker Node 與 Diver 處在相同的 JVM，并參考相同的原始 counter，這時候更新可能是正確的，但是在集群模式下一定不正確，所以在遇到此類問題時應優先使用累加器，

累加器的原理實際上很簡單：就是將每個副本變數的最終值傳回 Driver，由 Driver 聚合后得到最終值，并更新原始變數，

file

2.2 使用累加器

SparkContext 中定義了所有創建累加器的方法，需要注意的是：被中橫線劃掉的累加器方法在 Spark 2.0.0 之后被標識為廢棄，

使用示例和執行結果分別如下：

val data = https://www.cnblogs.com/hulichao/p/Array(1, 2, 3, 4, 5)
// 定義累加器
val accum = sc.longAccumulator("My Accumulator")
sc.parallelize(data).foreach(x => accum.add(x))
// 獲取累加器的值
accum.value

三、廣播變數

在上面介紹中閉包的程序中我們說道每個 Task 任務的閉包都會持有自由變數的副本，如果變數很大且 Task 任務很多的情況下，這必然會對網路 IO 造成壓力，為了解決這個情況，Spark 提供了廣播變數，

廣播變數的做法很簡單：就是不把副本變數分發到每個 Task 中，而是將其分發到每個 Executor，Executor 中的所有 Task 共享一個副本變數，

// 把一個陣列定義為一個廣播變數
val broadcastVar = sc.broadcast(Array(1, 2, 3, 4, 5))
// 之后用到該陣列時應優先使用廣播變數，而不是原值
sc.parallelize(broadcastVar.value).map(_ * 10).collect()