一、簡介

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

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

二、累加器

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

var counter = 0
val data = Array(1,2,3,4,5)
sc.parallelize(data).foreach(x => counter += x)
 println(counter)
複製程式碼
 

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

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 中的閉包

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

需要注意的是：在 Local 模式下，有可能執行 foreach

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

2.2 使用累加器

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

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

val data = Array(1,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,5))
// 之後用到該陣列時應優先使用廣播變數，而不是原值
sc.parallelize(broadcastVar.value).map(_ * 10).collect()
複製程式碼

參考資料

RDD Programming Guide

更多大資料系列文章可以參見 GitHub 開源專案： 大資料入門指南