上一篇部落格部落客已經為大家介紹了Spark中資料讀取與儲存，這一篇部落格則帶來了Spark中的程式設計進階。其中就涉及到了累加器與廣播變量的使用。
RDD程式設計進階
1.累加器
累加器用來對資訊進行聚合，通常在向 Spark傳遞函數時，比如使用 map() 函數或者用 filter() 傳條件時，可以使用驅動器程式中定義的變量，但是叢集中運作的每個任務都會得到這些變量的一份新的副本，更新這些副本的值也不會影響驅動器中的對應變量。如果我們想實作所有分片處理時更新共享變量的功能，那麼累加器可以實作我們想要的效果。
1.1系統累加器
針對一個輸入的日志檔案，如果我們想計算檔案中所有空行的數量，我們可以編寫以下程式：
scala> val notice = sc.textFile("./NOTICE")
notice: org.apache.spark.rdd.RDD[String] = ./NOTICE MapPartitionsRDD[40] at textFile at <console>:32
scala> val blanklines = sc.accumulator(0)
warning: there were two deprecation warnings; re-run with -deprecation for details
blanklines: org.apache.spark.Accumulator[Int] = 0
scala> val tmp = notice.flatMap(line => {
| if (line == "") {
| blanklines += 1
| }
| line.split(" ")
| })
tmp: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[41] at flatMap at <console>:36
scala> tmp.count()
res31: Long = 3213
scala> blanklines.value
res32: Int = 171 複制
通過在驅動器中調用SparkContext.accumulator(initialValue)方法，建立出存有初始值的累加器。傳回值為 org.apache.spark.Accumulator[T] 對象，其中 T 是初始值 initialValue 的類型。Spark閉包裡的執行器代碼可以使用累加器的 += 方法(在Java中是 add)增加累加器的值。 驅動器程式可以調用累加器的value屬性(在Java中使用value()或setValue())來通路累加器的值。
注意:工作節點上的任務不能通路累加器的值。從這些任務的角度來看，累加器是一個隻寫變量。
對于要在行動操作中使用的累加器，Spark隻會把每個任務對各累加器的修改應用一次。是以，如果想要一個無論在失敗還是重複計算時都絕對可靠的累加器，我們必須把它放在 foreach() 這樣的行動操作中。轉化操作中累加器可能會發生不止一次更新。
2.廣播變量(調優政策)
廣播變量用來高效分發較大的對象。向所有工作節點發送一個較大的隻讀值，以供一個或多個Spark操作使用。比如，如果你的應用需要向所有節點發送一個較大的隻讀查詢表，甚至是機器學習算法中的一個很大的特征向量，廣播變量用起來都很順手。 在多個并行操作中使用同一個變量，但是 Spark會為每個任務分别發送。
scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(35)
scala> broadcastVar.value
res33: Array[Int] = Array(1, 2, 3) 複制
使用廣播變量的過程如下：
(1) 通過對一個類型 T 的對象調用 SparkContext.broadcast 建立出一個 Broadcast[T] 對象。 任何可序列化的類型都可以這麼實作。
(2) 通過 value 屬性通路該對象的值(在 Java 中為 value() 方法)。
(3) 變量隻會被發到各個節點一次，應作為隻讀值處理(修改這個值不會影響到别的節點)。
本次的分享就到這裡，對大資料技術感興趣的朋友可以點贊關注一下喲~