天天看點

Spark 系列(六)—— 累加器與廣播變量

一、簡介

在 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

的 Worker Node 與 Diver 處在相同的 JVM,并引用相同的原始

counter

,這時候更新可能是正确的,但是在叢集模式下一定不正确。是以在遇到此類問題時應優先使用累加器。

累加器的原理實際上很簡單:就是将每個副本變量的最終值傳回 Driver,由 Driver 聚合後得到最終值,并更新原始變量。

2.2 使用累加器

SparkContext

中定義了所有建立累加器的方法,需要注意的是:被中橫線劃掉的累加器方法在 Spark 2.0.0 之後被辨別為廢棄。

Spark 系列(六)—— 累加器與廣播變量

使用示例和執行結果分别如下:

val data = Array(1, 2, 3, 4, 5)
// 定義累加器
val accum = sc.longAccumulator("My Accumulator")
sc.parallelize(data).foreach(x => accum.add(x))
// 擷取累加器的值
accum.value           
Spark 系列(六)—— 累加器與廣播變量

三、廣播變量

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

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

參考資料

繼續閱讀