spark中的閉包理解

官網：http://spark.apache.org/docs/2.3.0/rdd-programming-guide.html#understanding-closures-

---

 

Spark中一個很是難以理解的概念，就是在集羣中分佈式並行運行時操做的算子外部的變量的生命週期

一般來講，這個問題跟在RDD的算子中操做做用域外部的變量有關
所謂RDD算子中，操做做用域外部的變量，指的是，相似下面的語句:

int counter = 0;
JavaRDD<Integer> rdd = sc.parallelize(data);

// Wrong: Don't do this!!
rdd.foreach(x -> counter += x);

println("Counter value: " + counter);

此時，對rdd執行的foreach算子的做用域，其實僅僅是它的內部代碼，可是這裏卻操做了做用域外部的a變量

根據不一樣的編程語言的語法，這種功能是能夠作到的，而這種現象就叫作閉包

閉包簡單來講，就是操做的不屬於一個做用域範圍的變量

若是使用local模式運行spark做業，那麼實際只有一個jvm進程在執行這個做業
此時，你全部的RDD算子的代碼執行以及它們操做的外部變量，都是在一個進程的內存中，這個進程就是driver進程
此時是沒有任何問題的

可是在做業提交到集羣執行的模式下（不管是client或cluster模式，做業都是在集羣中運行的）
爲了分佈式並行執行你的做業，spark會將你的RDD算子操做，分散成多個task，放到集羣中的多個節點上的executor進程中去執行
每一個task執行的是相同的代碼，可是倒是處理不一樣的數據

在提交做業的task到集羣去執行以前，spark會先在driver端處理閉包
spark中的閉包，特指那些，不在算子的做用域內部，可是在做用域外部卻被算子處理和操做了的變量
而算子代碼的執行也須要這些變量才能順利執行
此時，這些閉包變量會被序列化成多個副本，而後每一個副本都發送到各個executor進程中，供那個executor進程運行的task執行代碼時使用

對於上面說的閉包變量處理機制
對於local模式，沒有任何特別的影響，畢竟都在一個jvm進程中，變量發送到executor，也不過就是進程中的一個線程而已
可是對於集羣運行模式來講，每一個executor進程，都會獲得一個閉包變量的副本，這個時候，就會出問題

所以閉包變量發送到executor進程中以後，就變成了一個一個獨立的變量副本了，這就是最關鍵的一點
此時在executor進程中，執行task和算子代碼時，訪問的閉包變量，也僅僅只是當前executor進程中的一個變量副本而已了
此時雖然在driver進程中，也有一個變量副本，可是卻徹底跟各個executor進程中的變量副本不是一個東西
此時，各個executor進程對於本身內存中的變量副本進行操做，即便改變了變量副本的值，可是對於driver端的程序，是徹底感知不到的
driver端的變量沒有被進行任何操做

所以綜上所述，在你使用集羣模式運行做業的時候，切忌不要在算子內部，對做用域外面的閉包變量進行改變其值的操做
由於那沒有任何意義，算子僅僅會在executor進程中，改變變量副本的值
對於driver端的變量沒有任何影響，咱們也獲取不到executor端的變量副本的值

若是但願在集羣模式下，對某個driver端的變量，進行分佈式並行地全局性的修改可使用Spark提供的Accumulator，全局累加器