前言

前面總結(jié)的幾篇spark踩坑博文中，我總結(jié)了自己在使用spark過(guò)程當(dāng)中踩過(guò)的一些坑和經(jīng)驗(yàn)。我們知道Spark是多機(jī)器集群部署的，分為Driver/Master/Worker，Master負(fù)責(zé)資源調(diào)度，Worker是不同的運(yùn)算節(jié)點(diǎn)，由Master統(tǒng)一調(diào)度。

而Driver是我們提交Spark程序的節(jié)點(diǎn)，并且所有的reduce類型的操作都會(huì)匯總到Driver節(jié)點(diǎn)進(jìn)行整合。節(jié)點(diǎn)之間會(huì)將map/reduce等操作函數(shù)傳遞一個(gè)獨(dú)立副本到每一個(gè)節(jié)點(diǎn)，這些變量也會(huì)復(fù)制到每臺(tái)機(jī)器上，而節(jié)點(diǎn)之間的運(yùn)算是相互獨(dú)立的，變量的更新并不會(huì)傳遞回Driver程序。

那么有個(gè)問題，如果我們想在節(jié)點(diǎn)之間共享一份變量，比如一份公共的配置項(xiàng)，該怎么辦呢?Spark為我們提供了兩種特定的共享變量，來(lái)完成節(jié)點(diǎn)間變量的共享。 本文首先簡(jiǎn)單的介紹spark以及spark streaming中累加器和廣播變量的使用方式，然后重點(diǎn)介紹一下如何更新廣播變量。

累加器

顧名思義，累加器是一種只能通過(guò)關(guān)聯(lián)操作進(jìn)行“加”操作的變量，因此它能夠高效的應(yīng)用于并行操作中。它們能夠用來(lái)實(shí)現(xiàn)counters和sums。Spark原生支持?jǐn)?shù)值類型的累加器，開發(fā)者可以自己添加支持的類型，在2.0.0之前的版本中，通過(guò)繼承AccumulatorParam來(lái)實(shí)現(xiàn)，而2.0.0之后的版本需要繼承AccumulatorV2來(lái)實(shí)現(xiàn)自定義類型的累加器。

如果創(chuàng)建了一個(gè)具名的累加器，它可以在spark的UI中顯示。這對(duì)于理解運(yùn)行階段(running stages)的過(guò)程有很重要的作用。如下圖：

在2.0.0之前版本中，累加器的聲明使用方式如下：

scala> val accum = sc.accumulator(0, "My Accumulator") 
accum: spark.Accumulator[Int] = 0 
 
scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x) 
... 
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s 
 
scala> accum.value 
res2: Int = 10 

累加器的聲明在2.0.0發(fā)生了變化，到2.1.0也有所變化，具體可以參考官方文檔，我們這里以2.1.0為例將代碼貼一下：

scala> val accum = sc.longAccumulator("My Accumulator") 
accum: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(My Accumulator), value: 0) 
 
scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x)) 
 
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s 
 
scala> accum.value 
res2: Long = 10 

廣播變量

累加器比較簡(jiǎn)單直觀，如果我們需要在spark中進(jìn)行一些全局統(tǒng)計(jì)就可以使用它。但是有時(shí)候僅僅一個(gè)累加器并不能滿足我們的需求，比如數(shù)據(jù)庫(kù)中一份公共配置表格，需要同步給各個(gè)節(jié)點(diǎn)進(jìn)行查詢。OK先來(lái)簡(jiǎn)單介紹下spark中的廣播變量：

廣播變量允許程序員緩存一個(gè)只讀的變量在每臺(tái)機(jī)器上面，而不是每個(gè)任務(wù)保存一份拷貝。例如，利用廣播變量，我們能夠以一種更有效率的方式將一個(gè)大數(shù)據(jù)量輸入集合的副本分配給每個(gè)節(jié)點(diǎn)。Spark也嘗試著利用有效的廣播算法去分配廣播變量，以減少通信的成本。

一個(gè)廣播變量可以通過(guò)調(diào)用SparkContext.broadcast(v)方法從一個(gè)初始變量v中創(chuàng)建。廣播變量是v的一個(gè)包裝變量，它的值可以通過(guò)value方法訪問，下面的代碼說(shuō)明了這個(gè)過(guò)程：

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3)) 
broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0) 
 
scala> broadcastVar.value 
res0: Array[Int] = Array(1, 2, 3) 

從上文我們可以看出廣播變量的聲明很簡(jiǎn)單，調(diào)用broadcast就能搞定，并且scala中一切可序列化的對(duì)象都是可以進(jìn)行廣播的，這就給了我們很大的想象空間，可以利用廣播變量將一些經(jīng)常訪問的大變量進(jìn)行廣播，而不是每個(gè)任務(wù)保存一份，這樣可以減少資源上的浪費(fèi)。

更新廣播變量(rebroadcast)

廣播變量可以用來(lái)更新一些大的配置變量，比如數(shù)據(jù)庫(kù)中的一張表格，那么有這樣一個(gè)問題，如果數(shù)據(jù)庫(kù)當(dāng)中的配置表格進(jìn)行了更新，我們需要重新廣播變量該怎么做呢。上文對(duì)廣播變量的說(shuō)明中，我們知道廣播變量是只讀的，也就是說(shuō)廣播出去的變量沒法再修改，那么我們應(yīng)該怎么解決這個(gè)問題呢?

答案是利用spark中的unpersist函數(shù)

Spark automatically monitors cache usage on each node and drops out old data partitions in a least-recently-used (LRU) fashion. If you would like to manually remove an RDD instead of waiting for it to fall out of the cache, use the RDD.unpersist() method.

上文是從spark官方文檔摘抄出來(lái)的，我們可以看出，正常來(lái)說(shuō)每個(gè)節(jié)點(diǎn)的數(shù)據(jù)是不需要我們操心的，spark會(huì)自動(dòng)按照LRU規(guī)則將老數(shù)據(jù)刪除，如果需要手動(dòng)刪除可以調(diào)用unpersist函數(shù)。

那么更新廣播變量的基本思路：將老的廣播變量刪除(unpersist)，然后重新廣播一遍新的廣播變量，為此簡(jiǎn)單包裝了一個(gè)用于廣播和更新廣播變量的wraper類，如下：

import java.io.{ ObjectInputStream, ObjectOutputStream } 
import org.apache.spark.broadcast.Broadcast 
import org.apache.spark.streaming.StreamingContext 
import scala.reflect.ClassTag 
 
// This wrapper lets us update brodcast variables within DStreams' foreachRDD 
// without running into serialization issues 
case class BroadcastWrapper[T: ClassTag]( 
    @transient private val ssc: StreamingContext, 
    @transient private val _v: T) { 
 
  @transient private var v = ssc.sparkContext.broadcast(_v) 
 
  def update(newValue: T, blocking: Boolean = false): Unit = { 
    // 刪除RDD是否需要鎖定 
    v.unpersist(blocking) 
    v = ssc.sparkContext.broadcast(newValue) 
  } 
 
  def value: T = v.value 
 
  private def writeObject(out: ObjectOutputStream): Unit = { 
    out.writeObject(v) 
  } 
 
  private def readObject(in: ObjectInputStream): Unit = { 
    v = in.readObject().asInstanceOf[Broadcast[T]] 
  } 
} 

利用該wrapper更新廣播變量，大致的處理邏輯如下：

// 定義 
val yourBroadcast = BroadcastWrapper[yourType](ssc, yourValue) 
 
yourStream.transform(rdd => { 
  //定期更新廣播變量 
  if (System.currentTimeMillis - someTime > Conf.updateFreq) { 
    yourBroadcast.update(newValue, true) 
  } 
  // do something else 
}) 

總結(jié)

spark中的共享變量是我們能夠在全局做出一些操作，比如record總數(shù)的統(tǒng)計(jì)更新，一些大變量配置項(xiàng)的廣播等等。而對(duì)于廣播變量，我們也可以監(jiān)控?cái)?shù)據(jù)庫(kù)中的變化，做到定時(shí)的重新廣播新的數(shù)據(jù)表配置情況，另外我使用上述方式，在每天***的數(shù)據(jù)實(shí)時(shí)流統(tǒng)計(jì)中表現(xiàn)穩(wěn)定，所以有相似問題的同學(xué)也可以進(jìn)行嘗試，有任何問題，歡迎隨時(shí)騷擾溝通。