作者：祝威廉

本文由 簡書 祝威廉 授權(quán)轉(zhuǎn)載，版權(quán)所有歸作者，轉(zhuǎn)載請聯(lián)系作者！

Facebook 60TB+級的Apache Spark應用案例 里大體有兩方面的PR，一個是Bug Fix，一個是性能優(yōu)化。這篇文章會對所有提及的Bug Issue進行一次解釋和說明。

前言

Facebook 60TB+級的Apache Spark應用案例,本來上周就準備看的，而且要求自己不能手機看，要在電腦上細細的看。然而終究是各種忙拖到了昨天晚上。

文章體現(xiàn)的工作，我覺得更像是一次挑戰(zhàn)賽，F(xiàn)acebook團隊通過層層加碼，最終將單個Spark Batch實例跑到了60T+ 的數(shù)據(jù)，這是一個了不起的成就，最最重要的是，他們完成這項挑戰(zhàn)賽后給社區(qū)帶來了三個好處：

• 在如此規(guī)模下，發(fā)現(xiàn)了一些Spark團隊以前很難發(fā)現(xiàn)的Bug
• 提交了大量的bug fix 和 new features,而且我們可以在Spark 1.6.2 /Spark 2.0 里享受到其中的成果
• 在如此規(guī)模下，我們也知道我們最可能遇到的一些問題。大體是OOM和Driver的限制。

說實在的，我覺得這篇文章，可以算是一篇工程論文了。而且只用了三個人力，不知道一共花了多久。

值得注意的是，大部分Bug都是和OOM相關(guān)的，這也是Spark的一個痛點，所以這次提交的PR質(zhì)量非常高。

Bug 剖析

Make PipedRDD robust to fetch failure SPARK-13793

這個Issue 還是比較明顯的。PipedRDD 在Task內(nèi)部啟動一個新的Java進程(假設我們叫做ChildProcessor)獲取數(shù)據(jù)。這里就會涉及到三個點：

1. 啟動一個線程往 ChildProcessor 寫數(shù)據(jù) (stdin writer)
2. 啟動一個線程監(jiān)控ChildProcessor的錯誤輸出 (stderr reader)
3. 獲取ChildProcessor輸入流，返回一個迭代器(Iterator)

既然都是讀取數(shù)據(jù)流，如果數(shù)據(jù)流因為某種異常原因關(guān)閉，那必然會拋出錯誤。所以我們需要記錄這個異常，對于1,2 兩個我們只要catch住異常，然后將異常記錄下來方便后續(xù)重新拋出。 那么什么時候拋出呢?迭代器有經(jīng)典的hasNext/next方法,每次hasNext時，我們都檢查下是否有Exception(來自1,2的)，如果有就拋出了。既然已經(jīng)異常了，我們就應該不需要繼續(xù)讀取這個分區(qū)的數(shù)據(jù)了。否則數(shù)據(jù)集很大的情況下，還要運行很長時間才能運行完。

在hasNext 為false的情況下，有兩類情況，一類是真的沒有數(shù)據(jù)了，一類是有異常了，比如有節(jié)點掛了，所以需要檢測下ChildProcessor的exitStatus狀態(tài)。如果不正常，就直接拋出異常，進行重試。

對于1，2兩點，原來都是沒有的，是這次Facebook團隊加上去的。

Configurable max number of fetch failures SPARK-13369

截止到我這篇文章發(fā)出，這個Issue 并沒有被接收。

我們知道，Shuffle 發(fā)生時，一般會發(fā)生有兩個Stage 產(chǎn)生，一個ShuffleMapStage (我們?nèi)∶麨?MapStage),他會寫入數(shù)據(jù)到文件中，接著下一個Stage (我們?nèi)∶麨镽educeStage) 就會去讀取對應的數(shù)據(jù)。 很多情況下，ReduceStage 去讀取數(shù)據(jù)MapStage 的數(shù)據(jù)會失敗，可能的原因比如有節(jié)點重啟導致MapStage產(chǎn)生的數(shù)據(jù)有丟失，此外還有GC超時等。這個時候Spark 就會重跑這兩個Stage,如果連續(xù)四次都發(fā)生這個問題，那么就會將整個Job給標記為失敗。 現(xiàn)階段(包括在剛發(fā)布的2.0)，這個數(shù)值是固定的，并不能夠設置。

@markhamstra 給出的質(zhì)疑是，如果發(fā)生節(jié)點失敗導致Stage 重新被Resubmit ,Resubmit后理論上不會再嘗試原來失敗的節(jié)點，如果連續(xù)四次都無法找到正常的階段運行這些任務，那么應該是有Bug，簡單增加重試次數(shù)雖然也有意義，但是治標不治本。

我個人認為在集群規(guī)模較大，任務較重的過程中，出現(xiàn)一個或者一批Node 掛掉啥的是很正常的，如果僅僅是因為某個Shuffle 導致整個Job失敗，對于那種大而耗時的任務顯然是不能接受的。個人認為應該講這個決定權(quán)交給用戶，也就是允許用戶配置嘗試次數(shù)。

Unresponsive driver SPARK-13279

這個Bug已經(jīng)在1.6.1, 2.0.0 中修復。 這個場景比較特殊，因為Facebook產(chǎn)生了高達200k的task數(shù)，原來給pendingTasksForExecutor:HashMap[String, ArrayBuffer[Int]] 添加新的task 的時候，都會根據(jù)Executor名獲取到已經(jīng)存在的列表，然后判斷該列表是否已經(jīng)包含了新Task,這個操作的時間復雜度是O(N^2)。在Task數(shù)比較小的情況下沒啥問題，但是一旦task數(shù)達到了200k,基本就要五分鐘，給人的感覺就是Driver沒啥反應了。

而且在實際運行任務的過程中，會通過一個特殊的dequeueTaskFromList結(jié)構(gòu)來排除掉已經(jīng)運行的任務，所以我們其實在addPendingTask 過程中不需要做這個檢測。

因為證明了沒有副作用，所以現(xiàn)在是沒啥問題了。但是我個人認為其實還有一種辦法是，取一個閾值，如果小于某個閾值則做double duplicate check,否則就直接加進去就好了。Spark 在很多地方也是這么做的。

這里對于那些Task數(shù)特別大的朋友有福了。

TimSort issue due to integer overflow for large buffer

該Bug在1.6.2, 2.0.0 已經(jīng)被解決。這個bug引起的問題現(xiàn)象初看起來會比較讓人費解，大體如下：

如圖所示似乎違反了簽名。其實問題本身確實比較復雜，通過提交了兩個patch 才解決了該問題。

一開始Facebook的哥們覺得應該是排序過程中內(nèi)存的數(shù)據(jù)(比如ShuffleExternalSorter等Sorter) 超過8G 引起的，所以限制了數(shù)量，大于一定數(shù)量之后就進行spill操作。 后面一個新的PR應該是發(fā)現(xiàn)了問題的根源，在UnsafeSortDataFormat.copyRange() 和ShuffleSortDataFormat copyRange() 里，里面數(shù)組的偏移量是Integer類型，雖然數(shù)據(jù)集的大小不至于超過Int的最大值，但是在特定數(shù)據(jù)分布下且數(shù)據(jù)集>268.43 million 并則會觸發(fā)這個Bug。我看了下，原先 Platform.copyMemory 簽名本身也是Long的，但是實現(xiàn)copyRange的時候，默認傳進去的是Int,所以產(chǎn)生了這個問題。大家瞅一眼代碼就知道了。

Fix Spark executor OOM

該Bug 也是在1.6.2, 2.0.0 被修正。

這個問題是這樣的，Spark MemoryManager 可能認為還有10M內(nèi)存，但是此時實際JVM可以提供給MemroyManager的內(nèi)存只有5M了。所以分配內(nèi)存的時候，就拋OOM了。這個時候應該捕獲該OOM,并且保留已經(jīng)申請到內(nèi)存不歸還，讓MemoryManger 以為內(nèi)存不夠了，然后進行splill操作，從而湊足需要的內(nèi)存。我們看TaskMemoryManager.allocatePage 方法。

如果發(fā)生OOM了，則會捕獲一次，,并且通過acquiredButNotUsed記住已經(jīng)申請的量，最后再次調(diào)用allocatePage。這個時候allocatePage里的acquireExecutionMemory 方法可能發(fā)現(xiàn)自己內(nèi)存不足了，就會發(fā)生spill了，從而釋放出內(nèi)存。

其實這之前的代碼也考慮過，但是沒有在allocatePage的層次上做。這個Bug估計在單個Executor 并行運行Task數(shù)比較多的時候比較嚴重和容易發(fā)生的。

Fix memory leak in the sorter SPARK-14363

這個Bug 也是在1.6.2, 2.0.0被修正。

在Spark排序中，指針和數(shù)據(jù)時分開存儲的，進行spill操作其實是把數(shù)據(jù)替換到磁盤上。但是指針數(shù)組是必須在內(nèi)存里。當數(shù)據(jù)被spill后，相應的，指向這些記錄的指針其實也是要被釋放的。數(shù)據(jù)量很大的時候，指針數(shù)組的大小也很可觀。而且有一點值得指出的是，比如某個Executor 有五個Task并行運行，如果其中有三個完成了，那么可用內(nèi)存增大，緩存到內(nèi)存的數(shù)據(jù)就會變多，這個時候剩下的兩個Task的指針數(shù)組也會增大，從而占用更多內(nèi)存，接著新運行的三個Task可用內(nèi)存變小了，從而失去了公平性。

這些各個Sorter里都需要修正。

紅框部分便是釋放指針數(shù)組的地方。里面會重新按初始initialSize值申請一塊指針數(shù)組的內(nèi)存。

原文>>>

End.