• Spark踩坑記——初試

• Spark踩坑記——數(shù)據(jù)庫（Hbase+Mysql）

• Spark踩坑記——Spark Streaming+kafka應用及調優(yōu)
  在前面總結的幾篇spark踩坑博文中，我總結了自己在使用spark過程當中踩過的一些坑和經驗。我們知道Spark是多機器集群部署的，分為Driver/Master/Worker，Master負責資源調度，Worker是不同的運算節(jié)點，由Master統(tǒng)一調度，而Driver是我們提交Spark程序的節(jié)點，并且所有的reduce類型的操作都會匯總到Driver節(jié)點進行整合。節(jié)點之間會將map/reduce等操作函數(shù)傳遞一個獨立副本到每一個節(jié)點，這些變量也會復制到每臺機器上，而節(jié)點之間的運算是相互獨立的，變量的更新并不會傳遞回Driver程序。那么有個問題，如果我們想在節(jié)點之間共享一份變量，比如一份公共的配置項，該怎么辦呢？Spark為我們提供了兩種特定的共享變量，來完成節(jié)點間變量的共享。
  本文首先簡單的介紹spark以及spark streaming中累加器和廣播變量的使用方式，然后重點介紹一下如何更新廣播變量。

累加器

顧名思義，累加器是一種只能通過關聯(lián)操作進行“加”操作的變量，因此它能夠高效的應用于并行操作中。它們能夠用來實現(xiàn)counters和sums。Spark原生支持數(shù)值類型的累加器，開發(fā)者可以自己添加支持的類型，在2.0.0之前的版本中，通過繼承AccumulatorParam來實現(xiàn)，而2.0.0之后的版本需要繼承AccumulatorV2來實現(xiàn)自定義類型的累加器。
如果創(chuàng)建了一個具名的累加器，它可以在spark的UI中顯示。這對于理解運行階段(running stages)的過程有很重要的作用。如下圖：

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

scala> val accum = sc.accumulator(0, "My Accumulator")accum: spark.Accumulator[Int] = 0scala> 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.valueres2: 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
        
		

        		
網友評論
     		
				

			
			
			
		
		
		
    	
    	
        	

        		
更多精彩分享
        		

        			
        		
        			
	        		
	        		學習是年輕人改變自己的最好方式