Spark學習——資料傾斜
資料傾斜是一種很常見的問題（依據二八定律），簡單來說，比方WordCount中某個Key對應的資料量非常大的話，就會產生資料傾斜，導致兩個後果：
- OOM（單或少數的節點）；
- 拖慢整個Job執行時間（其他已經完成的節點都在等這個還在做的節點）。
2. 解決資料傾斜需要
搞定 Shuffle； 搞定業務場景； 搞定 CPU core 的使用情況； 搞定 OOM 的根本原因等：一般都因為資料傾斜（某task任務的資料量過大，GC壓力大，和Kafka不同在於Kafka的記憶體不經過JVM，其基於Linux的Page）。
3. 導致Spark資料傾斜的本質
Shuffle時，需將各節點的相同key的資料拉取到某節點上的一個task來處理，若某個key對應的資料量很大就會發生資料傾斜。比方說大部分key對應10條資料，某key對應10萬條，大部分task只會被分配10條資料，很快做完，個別task分配10萬條資料，不僅執行時間長，且整個stage的作業時間由最慢的task決定。
資料傾斜只會發生在Shuffle過程，以下演算法可能觸發Shuffle操作： distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition等。
4. 定位最慢的Task所處的原始碼位置
步驟一： 看資料傾斜發生在哪個stage（也就是看以上運算元出現在哪個階段）。yarn-client模式下檢視本地log或Spark Web UI中當前執行的是哪個stage；yarn-cluster模式下，通過Spark Web UI檢視執行到了哪個Stage。 主要看最慢的Stage各task分配的資料量，來確定是否是資料傾斜。
步驟二：根據Stage劃分，推算傾斜發生的程式碼（必然有Shuffle類運算元）。簡單實用方法：只要看到shuffle類運算元或Spark SQL的SQL語句會有Shuffle類的運算元的句子，就可以該地方劃分為前後兩個Stage。（之前用Python的PySpark介面，Spark Web UI會檢視task在原始碼中的行數，Java或者Scala雖沒用過，但我想應該有）
5. 解決方案
方案一：使用Hive ETL預處理
場景：若Hive表中資料不均勻，且業務中會頻繁用Spark對Hive表分析； 思路：用Hive對資料預處理（對key聚合等操作），原本是Spark對Hive的原表操作，現在就是對Hive預處理後的表操作； 原理：從根源解決了資料傾斜，規避了了Spark進行Shuffle類運算元操作。但Hive ETL中進行聚合等操作會發生資料傾斜，只是把慢轉移給了Hive ETL； 優點：方便，效果好，規避了Spark資料傾斜； 缺點：治標不治本，Hive ETL會資料傾斜。
方案二：過濾導致傾斜的key
場景：發生傾斜的key很少且不重要； 思路：對發生傾斜的key過濾掉。比方在Spark SQL中用where子句或filter過濾，若每次作業執行，需要動態判定可使用sample運算元對RDD取樣後取資料量最多的key過濾； 原理：對傾斜的key過濾後，這些key便不會參與後面的計算，從本質上消除資料傾斜； 優點：簡單，效果明顯； 缺點：適用場景少，實際中導致傾斜的key很多。
方案三：提高Shuffle操作並行度
場景：任何場景都可以，優先選擇的最簡單方案； 思路：對RDD操作的Shuffle運算元傳入一個引數，也就是設定Shuffle運算元執行時的Shuffle read task數量。對於Spark SQL的Shuffle類語句（如group by，join）即spark.sql.shuffle.partitions，代表shuffle read task的並行度，預設值是200可修改； 原理：增大shuffle read task引數值，讓每個task處理比原來更少的資料； 優點：簡單，有效； 缺點：緩解的效果很有限。
方案四：兩階段聚合（區域性聚合+全域性聚合）
場景：對RDD進行reduceByKey等聚合類shuffle運算元，SparkSQL的groupBy做分組聚合這兩種情況 思路：首先通過map給每個key打上n以內的隨機數的字首並進行區域性聚合，即(hello, 1) (hello, 1) (hello, 1) (hello, 1)變為(1_hello, 1) (1_hello, 1) (2_hello, 1)，並進行reduceByKey的區域性聚合，然後再次map將key的字首隨機數去掉再次進行全域性聚合； 原理：對原本相同的key進行隨機數附加，變成不同key，讓原本一個task處理的資料分攤到多個task做區域性聚合，規避單task資料過量。之後再去隨機字首進行全域性聚合； 優點：效果非常好（對聚合類Shuffle操作的傾斜問題）； 缺點：範圍窄（僅適用於聚合類的Shuffle操作，join類的Shuffle還需其它方案）。
方案五：將reduce join轉為map join
場景：對RDD或Spark SQL使用join類操作或語句，且join操作的RDD或表比較小（百兆或1,2G）； 思路：使用broadcast和map類運算元實現join的功能替代原本的join，徹底規避shuffle。對較小RDD直接collect到記憶體，並建立broadcast變數；並對另外一個RDD執行map類運算元，在該運算元的函式中，從broadcast變數（collect出的較小RDD）與當前RDD中的每條資料依次比對key，相同的key執行你需要方式的join； 原理：若RDD較小，可採用廣播小的RDD，並對大的RDD進行map，來實現與join同樣的效果。簡而言之，用broadcast-map代替join，規避join帶來的shuffle（無Shuffle無傾斜）； 優點：效果很好（對join操作導致的傾斜），根治； 缺點：適用場景小（大表+小表），廣播（driver和executor節點都會駐留小表資料）小表也耗記憶體。
方案六：取樣傾斜key並分拆join操作
場景：兩個較大的（無法採用方案五）RDD/Hive表進行join時，且一個RDD/Hive表中少數key資料量過大，另一個RDD/Hive表的key分佈較均勻（RDD中兩者之一有一個更傾斜）； 思路：
- 對更傾斜rdd1進行取樣（RDD.sample）並統計出資料量最大的幾個key；
- 對這幾個傾斜的key從原本rdd1中拆出形成一個單獨的rdd1_1，並打上0~n的隨機數字首，被拆分的原rdd1的另一部分（不包含傾斜key）又形成一個新rdd1_2；
- 對rdd2過濾出rdd1傾斜的key，得到rdd2_1，並將其中每條資料擴n倍，對每條資料按順序附加0~n的字首，被拆分出key的rdd2也獨立形成另一個rdd2_2； 【個人認為，這裡擴了n倍，最後union完還需要將每個傾斜key對應的value減去(n-1)】
- 將加了隨機字首的rdd1_1和rdd2_1進行join（此時原本傾斜的key被打散n份並被分散到更多的task中進行join）； 【個人認為，這裡應該做兩次join，兩次join中間有一個map去字首】
- 另外兩個普通的RDD（rdd1_2、rdd2_2）照常join；
- 最後將兩次join的結果用union結合得到最終的join結果。 原理：對join導致的傾斜是因為某幾個key，可將原本RDD中的傾斜key拆分出原RDD得到新RDD，並以加隨機字首的方式打散n份做join，將傾斜key對應的大量資料分攤到更多task上來規避傾斜； 優點：前提是join導致的傾斜（某幾個key傾斜），避免佔用過多記憶體（只需對少數傾斜key擴容n倍）； 缺點：對過多傾斜key不適用。