何謂數(shù)據(jù)傾斜?數(shù)據(jù)傾斜指的是并行處理的數(shù)據(jù)集中，某一部分(如Spark或Kafka的一個Partition)的數(shù)據(jù)顯著多于其它部分，從而使得該部分的處理速度成為整個數(shù)據(jù)集處理的瓶頸。
一、數(shù)據(jù)傾斜概述
1.1 什么是數(shù)據(jù)傾斜
對Hadoop、Spark、Flink這樣的大數(shù)據(jù)系統(tǒng)來講，數(shù)據(jù)量大并不可怕，可怕的是數(shù)據(jù)傾斜。
何謂數(shù)據(jù)傾斜?數(shù)據(jù)傾斜指的是，并行處理的數(shù)據(jù)集中，某一部分(如Spark或Kafka的一個Partition)的數(shù)據(jù)顯著多于其它部分，從而使得該部分的處理速度成為整個數(shù)據(jù)集處理的瓶頸。
對于分布式系統(tǒng)而言，理想情況下，隨著系統(tǒng)規(guī)模(節(jié)點數(shù)量)的增加，應用整體耗時線性下降。如果一臺機器處理一批大量數(shù)據(jù)需要120分鐘，當機器數(shù)量增加到三時，理想的耗時為120 / 3 = 40分鐘，如下圖所示
但是，上述情況只是理想情況，實際上將單機任務轉換成分布式任務后，會有overhead，使得總的任務量較之單機時有所增加，所以每臺機器的執(zhí)行時間加起來比單臺機器時更大。這里暫不考慮這些overhead，假設單機任務轉換成分布式任務后，總任務量不變。
但即使如此，想做到分布式情況下每臺機器執(zhí)行時間是單機時的1 / N，就必須保證每臺機器的任務量相等。不幸的是，很多時候，任務的分配是不均勻的，甚至不均勻到大部分任務被分配到個別機器上，其它大部分機器所分配的任務量只占總得的小部分。比如一臺機器負責處理80%的任務，另外兩臺機器各處理10%的任務，如下圖所示
1.2 數(shù)據(jù)傾斜發(fā)生時的現(xiàn)象
• 絕大多數(shù) task 執(zhí)行得都非常快，但個別 task 執(zhí)行極慢。比如，總共有 1000 個 task，997 個 task 都在 1 分鐘之內執(zhí)行完了，但是剩余兩三個 task 卻要一兩個小時。這種情況很常見。
• 原本能夠正常執(zhí)行的 Spark 作業(yè)，某天突然報出 OOM(內存溢出)異常，觀察異常棧，是我們寫 的業(yè)務代碼造成的。這種情況比較少見。
• Task 類似下圖所示
總結:
1. 大部分任務都很快執(zhí)行完，用時也相差無幾，但個別Task執(zhí)行耗時很長，整個應用程序一直處于99%左右 的狀態(tài)。
2. 一直運行正常的Spark Application昨晚突然OOM了。
1.3 數(shù)據(jù)傾斜發(fā)生的原理
數(shù)據(jù)傾斜的原理很簡單: 在進行 shuffle 的時候，必須將各個節(jié)點上相同的 key 的數(shù)據(jù)拉取到某個節(jié)點 上的一個 task 來進行處理，比如按照 key 進行聚合或 join 等操作。此時如果某個 key 對應的數(shù)據(jù)量特 別大的話，就會發(fā)生數(shù)據(jù)傾斜。比如大部分 key 對應 10 條數(shù)據(jù)，但是個別 key 卻對應了 100 萬條數(shù) 據(jù)，那么大部分 task 可能就只會分配到 10 條數(shù)據(jù)，然后 1 秒鐘就運行完了;但是個別 task 可能分配 到了 100 萬數(shù)據(jù)，要運行一兩個小時。因此，整個 Spark 作業(yè)的運行進度是由運行時間最長的那個 task 決定的。
因此出現(xiàn)數(shù)據(jù)傾斜的時候，Spark 作業(yè)看起來會運行得非常緩慢，甚至可能因為某個 task 處理的數(shù)據(jù) 量過大導致內存溢出。
下圖就是一個很清晰的例子:hello 這個 key，在三個節(jié)點上對應了總共 7 條數(shù)據(jù)，這些數(shù)據(jù)都會被拉 取到同一個task中進行處理;而world 和 you 這兩個 key 分別才對應 1 條數(shù)據(jù)，所以另外兩個 task 只 要分別處理 1 條數(shù)據(jù)即可。此時第一個 task 的運行時間可能是另外兩個 task 的 7 倍，而整個 stage 的 運行速度也由運行最慢的那個 task 所決定。
總結：
* 數(shù)據(jù)傾斜發(fā)生的本質，就是在執(zhí)行多階段的計算的時候，中間的shuffle策略可能導致分發(fā)到下 游Task的數(shù)據(jù)量不均勻，進而導致下游Task執(zhí)行時長的不一致。不完全均勻是正常的，但是如果相差太大，那么就產生性能問題了。
1.4 數(shù)據(jù)傾斜的危害
從上圖可見，當出現(xiàn)數(shù)據(jù)傾斜時，小量任務耗時遠高于其它任務，從而使得整體耗時過大，未能充分發(fā) 揮分布式系統(tǒng)的并行計算優(yōu)勢。另外，當發(fā)生數(shù)據(jù)傾斜時，
少量部分任務處理的數(shù)據(jù)量過大，可能造成 內存不足使得任務失敗，并進而引進整個應用失敗。如果應用并沒有因此失敗，但是大量正常任務都早 早完成處于等待狀態(tài)，資源得不到充分利用。
總結：
1. 整體耗時過大(整個任務的完成由執(zhí)行時間最長的那個Task決定)
2. 應用程序可能異常退出(某個Task執(zhí)行時處理的數(shù)據(jù)量遠遠大于正常節(jié)點，則需要的資源容易出現(xiàn)瓶頸， 當資源不足，則應用程序退出)
3. 資源閑置(處理等待狀態(tài)的Task資源得不到及時的釋放，處于閑置浪費狀態(tài))
1.5 數(shù)據(jù)傾斜是如何造成的
在 Spark 中，同一個 Stage 的不同 Partition 可以并行處理，而具有依賴關系的不同 Stage 之間是串行 處理的。假設某個 Spark Job 分為Stage0 和 Stage1 兩個 Stage，且 Stage1 依賴于 Stage0，那 Stage0 完全處理結束之前不會處理 Stage1。而 Stage0 可能包含 N 個Task，這 N 個 Task 可以并行進行。如 果其中 N-1 個 Task 都在 10 秒內完成，而另外一個 Task 卻耗時 1 分鐘，那該 Stage 的總時間至少為 1 分鐘。換句話說，一個 Stage 所耗費的時間，主要由最慢的那個 Task 決定。由于同一個 Stage 內的所有 Task 執(zhí)行相同的計算，在排除不同計算節(jié)點計算能力差異的前提下，不同 Task 之間耗時的差異主要由該 Task 所處理的數(shù)據(jù)量決定。Stage 的數(shù)據(jù)來源主要分為如下兩類:
1. 數(shù)據(jù)源本身分布有問題:從數(shù)據(jù)源直接讀取。如讀取HDFS，Kafka，有可能出現(xiàn)，大概率不會
2. 自己指定的分區(qū)規(guī)則:讀取上一個 Stage 的 Shuffle 數(shù)據(jù)
樸素的分布式計算的核心思想:
1. 大問題拆分成小問題:分而治之
2. 既然要分開算，那最后就一定要把分開計算的那么多的小 Task 的結果執(zhí)行匯總
3. 所以必然分布式計算引擎的設計中，應用程序的執(zhí)行一定是分階段
4. 分布計算引擎的而核心:一個復雜的分布式計算應用程序的執(zhí)行肯定要分成多個階段，每個階段分布式并 行運行多個Task
5. DAG引擎:
Spark: stage1 ==> stage2 ===> stage3
mapreduce: 就只有兩個階段:mapper reducer
階段與階段之間需要進行 shuffle，只要進行了數(shù)據(jù)混洗，就存在著數(shù)據(jù)分發(fā)不均勻的情況。如果情況嚴 重，就是數(shù)據(jù)傾斜。
分布式計算引擎的設計，免不了有shuffle，既然有shuffle操作，就一定有產生數(shù)據(jù)傾斜的可能。如果 你是做大數(shù)據(jù)處理的，就一定會遇到 數(shù)據(jù)傾斜!
京公網安備 11010802030320號