[[139085]]

是的，Spark很快。但是它不保證它算出的值是對(duì)的，哪怕你要做的只是簡(jiǎn)單的整數(shù)累加。

Spark***的一篇論文是：《Spark: Cluster Computing with Working Sets》。當(dāng)你讀它的時(shí)候你需要明白：文中代碼不保證計(jì)算結(jié)果是正確的。具體來說，它的Logistic Regression的代碼在map階段用到了accumulator。下面解釋為什么這么做是錯(cuò)誤的。

假設(shè)有這樣一個(gè)簡(jiǎn)單的任務(wù)：

input file的每一行是100個(gè)整數(shù)，要求豎著加下來

例如：

輸入

1 2 3 4 5 ... 100

1 2 3 4 5 ... 200

1 3 3 4 5 ... 100

輸出

3 7 9 12 15 ... 400

很簡(jiǎn)單，對(duì)吧?是個(gè)豬都會(huì)算。在hadoop上這個(gè)問題可以通過Map reduce來解決。首先把輸入文件分成N個(gè)大小相等的塊。然后每個(gè)塊輸出一行100個(gè)整數(shù)，如 2 4 6 8 10 ... 200

然后reducer接收每個(gè)mapper的輸出結(jié)果，累加起來得到最終結(jié)果。

缺點(diǎn)是： 從mapper到reducer是需要DISK-IO及網(wǎng)絡(luò)傳輸?shù)摹Ｄ敲葱枰獋鬏擭*100個(gè)整數(shù)。當(dāng)輸入集的維數(shù)很大(每行有上百萬個(gè)字節(jié))的時(shí)候，很浪費(fèi)。

spark很巧妙的引入了accumulator的概念。同一臺(tái)機(jī)器上所有的task的輸出，會(huì)先在這個(gè)機(jī)器上進(jìn)行本地匯總，然后再發(fā)給 reducer。這樣就不再是task數(shù)量*維數(shù)，而是機(jī)器數(shù)量*維數(shù)。會(huì)節(jié)省不少。具體來說，在做機(jī)器學(xué)習(xí)的時(shí)候，大家很習(xí)慣的用 accumulator來做這樣的計(jì)算。

accumulator是被很careful設(shè)計(jì)的。比如，只有master節(jié)點(diǎn)能讀取accumulator的值，worker節(jié)點(diǎn)不能。在“Performance and Scalability of Broadcast in Spark

”一文中，作者寫到：“Accumulators can be defined for any type that has an “add” operation and a “zero” value. Due to their “add-only” semantics, they are easy to make fault-tolerant.” 。但真的是這樣嗎?并不是。

accumulator如果不是運(yùn)行在運(yùn)算的***一環(huán)，那么正確性無法保證。因?yàn)閍ccumulator不是map/reduce函數(shù)的輸入或輸出，accumulator是表達(dá)式求值中的side-effect。舉個(gè)例子：

val acc = sc.accumulator(0) 
 
data.map(x => acc += 1; f(x)) 
 
data.count() 
 
// acc should equal data.count() here 
 
data.foreach{...} 
 
// Now, acc = 2 * data.count() because the map() was recomputed. 

這個(gè)問題被spark的創(chuàng)始人Matei標(biāo)為Won't Fix。

那么是不是寫代碼小心點(diǎn)不要觸發(fā)重復(fù)計(jì)算就行了呢?也不是。task是有可能fail-retry的，再或者因?yàn)槟骋粋€(gè)task執(zhí)行的慢，所以同時(shí)有它的多個(gè)副本在跑。這些都可能會(huì)導(dǎo)致accumulator結(jié)果不正確。 Accumulators只能用在RDD的actions中，不能用在Transformations。舉例來說：可以在reduce函數(shù)中用，但是不能在map函數(shù)中用。

如果不用accumlators，但又想節(jié)省網(wǎng)絡(luò)傳輸，那么Matei說：“I would suggest creating fewer tasks. If your input file has a lot of blocks and hence a lot of parallel tasks, you can use CoalescedRDD to create an RDD with fewer blocks from it. ”

意思就是說，那你就把task劃分大一點(diǎn)，把task的數(shù)量減少。比如每臺(tái)機(jī)器只有1個(gè)task。 Downside其實(shí)也很明顯，任務(wù)的執(zhí)行容易不balance。

參考： https://issues.apache.org/jira/browse/SPARK-732

https://issues.apache.org/jira/browse/SPARK-3628

https://issues.apache.org/jira/browse/SPARK-5490

https://github.com/apache/spark/pull/228

原文鏈接：http://www.sunchangming.com/blog/post/4672.html