數據傾斜那些坑 你都邁過去了嗎？
0x00 前言
數據傾斜是大數據領域繞不開的攔路虎，當你所需處理的數據量到達了上億甚至是千億條的時候，數據傾斜將是橫在你面前一道巨大的坎。
邁的過去，將會海闊天空!邁不過去，就要做好準備：很可能有幾周甚至幾月都要頭疼于數據傾斜導致的各類詭異的問題。
文章結構
- 先大致解釋一下什么是數據傾斜
- 再根據幾個場景來描述一下數據傾斜產生的情況
- 詳細分析一下在Hadoop和Spark中產生數據傾斜的原因
- 如何解決(優化)數據傾斜問題?
0x01 什么是數據傾斜
簡單的講，數據傾斜就是我們在計算數據的時候，數據的分散度不夠，導致大量的數據集中到了一臺或者幾臺機器上計算，這些數據的計算速度遠遠低于平均計算速度，導致整個計算過程過慢。
一、關鍵字：數據傾斜
相信大部分做數據的童鞋們都會遇到數據傾斜，數據傾斜會發生在數據開發的各個環節中，比如：
- 用 Hive 算數據的時候 reduce 階段卡在 99.99%
- 用 Spark Streaming 做實時算法時候，一直會有 executor 出現 OOM 的錯誤，但是其余的 executor 內存使用率卻很低。
這些問題經常會困擾我們，辛辛苦苦等了幾個小時的數據就是跑不出來，心里多難過啊。
例子很多，這里先隨便舉兩個，后文會詳細的說明。
二、關鍵字：千億級
為什么要突出這么大數據量?先說一下筆者自己最初對數據量的理解：
數據量大就了不起了?數據量少，機器也少，計算能力也是有限的，因此難度也是一樣的。憑什么數據量大就會有數據傾斜，數據量小就沒有?
這樣理解也有道理，但是比較片面，舉兩個場景來對比：
- 公司一：總用戶量 1000 萬，5 臺 64G 內存的的服務器。
- 公司二：總用戶量 10 億，1000 臺 64G 內存的服務器。
兩個公司都部署了 Hadoop 集群。假設現在遇到了數據傾斜，發生什么?
公司一的數據分時童鞋在做 join 的時候發生了數據傾斜，會導致有幾百萬用戶的相關數據集中到了一臺服務器上，幾百萬的用戶數據，說大也不大，正常字段量的數據的話 64G 還是能輕松處理掉的。
公司二的數據分時童鞋在做 join的時候也發生了數據傾斜，可能會有 1 億的用戶相關數據集中到了一臺機器上了(相信我，這很常見)，這時候一臺機器就很難搞定了，***會很難算出結果。
0x02 數據傾斜長什么樣
工作中遇到的大部分的數據傾斜問題都解決了，而且也不想重新運行任務來截圖，下面會分幾個場景來描述一下數據傾斜的特征，方便讀者辨別。
由于 Hadoop 和 Spark 是最常見的兩個計算平臺，下面就以這兩個平臺說明：
一、Hadoop中的數據傾斜
Hadoop 中最常用的是的是 Mapreduce 和 Hive ，雖說 Hive ***也是用 MR 來執行(至少目前Hive內存計算并不普及)，但是畢竟寫的內容邏輯區別很大，一個是程序，一個是 Sql，因此這里稍作區分。
Hadoop中的數據傾斜主要表現在、ruduce 階段卡在 99.99%，一直 99.99% 不能結束。
這里如果詳細的看日志或者和監控界面的話會發現：
- 有一個多幾個 reduce 卡住
- 各種 container 報錯 OOM
- 讀寫的數據量極大，至少遠遠超過其它正常的 reduce
- 伴隨著數據傾斜，會出現任務被 kill 等各種詭異的表現。
經驗：Hive 的數據傾斜，一般都發生在 Sql 中 Group 和 On 上，而且和數據邏輯綁定比較深。
二、Spark 中的數據傾斜
Spark 中的數據傾斜也很常見，這里包括Spark Streaming 和 Spark Sql，表現主要有下面幾種：
- Executor lost，OOM，Shuffle過程出錯
- Driver OOM
- 單個 executor 執行時間特別久，整體任務卡在某個階段不能結束
- 正常運行的任務突然失敗
補充一下，在 Spark streaming 程序中，數據傾斜更容易出現，特別是在程序中包含一些類似 Sql 的 join、group 這種操作的時候。 因為 Spark Streaming 程序在運行的時候，我們一般不會分配特別多的內存，因此一旦在這個過程中出現一些數據傾斜，就十分容易造成 OOM。
0x03 數據傾斜的原理
一、數據傾斜產生的原因
我們以 Spark 和 Hive 的使用場景為例。他們在做數據運算的時候會涉及到，count distinct、group by、join 等操作，這些都會觸發 Shuffle 動作，一旦觸發，所有相同 key 的值就會拉到一個或幾個節點上，就容易發生單點問題。
二、萬惡的 Shuffle
Shuffle 是一個能產生奇跡的地方，不管是在 Spark 還是 Hadoop 中，它們的作用都是至關重要的。關于 Shuffle 的原理，這里不再講述，看看 Hadoop 相關的論文或者文章理解一下就 ok。這里主要針對在 Shuffle 如何產生了數據傾斜。
Hadoop 和 Spark 在 Shuffle 過程中產生數據傾斜的原理基本類似。
大部分數據傾斜的原理就類似于下圖，很明了，因為數據分布不均勻，導致大量的數據分配到了一個節點。
三、從數據角度來理解數據傾斜
我們舉一個例子，就說數據默認值的設計吧，假設我們有兩張表：
- user(用戶信息表)：userid，register_ip
- ip(IP表)：ip，register_user_cnt
這可能是兩個不同的人開發的數據表，如果我們的數據規范不太完善的話，會出現一種情況，user 表中的 register_ip 字段，如果獲取不到這個信息，我們默認為 null，但是在 ip 表中，我們在統計這個值的時候，為了方便，我們把獲取不到 ip 的用戶，統一認為他們的 ip 為 0。
兩邊其實都沒有錯的，但是一旦我們做關聯了會出現什么情況，這個任務會在做關聯的階段，也就是sql的on的階段卡死。
四、從業務計角度來理解數據傾斜
數據往往和業務是強相關的，業務的場景直接影響到了數據的分布。
再舉一個例子，比如就說訂單場景吧，我們在某一天在北京和上海兩個城市多了強力的推廣，結果可能是這兩個城市的訂單量增長了10000%，其余城市的數據量不變。
然后我們要統計不同城市的訂單情況，這樣，一做 group 操作，可能直接就數據傾斜了。
0x04 如何解決
數據傾斜的產生是有一些討論的，解決它們也是有一些討論的，本章會先給出幾個解決數據傾斜的思路，然后對 Hadoop 和 Spark 分別給出一些解決數據傾斜的方案。
注意： 很多數據傾斜的問題，都可以用和平臺無關的方式解決，比如更好的數據預處理， 異常值的過濾等，因此筆者認為，解決數據傾斜的重點在于對數據設計和業務的理解，這兩個搞清楚了，數據傾斜就解決了大部分了。
一、幾個思路
解決數據傾斜有這幾個思路：
- 業務邏輯，我們從業務邏輯的層面上來優化數據傾斜，比如上面的例子，我們單獨對這兩個城市來做 count，***和其它城市做整合。
- 程序層面，比如說在 Hive 中，經常遇到count(distinct)操作，這樣會導致最終只有一個 reduce，我們可以先 group 再在外面包一層 count，就可以了。
- 調參方面，Hadoop 和 Spark 都自帶了很多的參數和機制來調節數據傾斜，合理利用它們就能解決大部分問題。
二、從業務和數據上解決數據傾斜
很多數據傾斜都是在數據的使用上造成的。我們舉幾個場景，并分別給出它們的解決方案。
數據分布不均勻：
- 前面提到的“從數據角度來理解數據傾斜”和“從業務計角度來理解數據傾斜”中的例子，其實都是數據分布不均勻的類型，這種情況和計算平臺無關，我們能通過設計的角度嘗試解決它。
有損的方法：
- 找到異常數據，比如 ip 為 0 的數據，過濾掉
無損的方法：
- 對分布不均勻的數據，單獨計算
- 先對 key 做一層 hash，先將數據打散讓它的并行度變大，再匯集
數據預處理
三、Hadoop平臺的優化方法
列出來一些方法和思路，具體的參數和用法在官網看就行了。
- map join 方式
- count distinct 的操作，先轉成 group，再 count
- ***膏藥：hive.groupby.skewindata=true
- left semi join的使用
- 設置 map 端輸出、中間結果壓縮。(不完全是解決數據傾斜的問題，但是減少了 IO 讀寫和網絡傳輸，能提高很多效率)
四、Spark平臺的優化方法
列出來一些方法和思路，具體的參數和用法在官網看就行了。
- mapjoin 方式
- 設置 rdd 壓縮
- 合理設置 driver 的內存
- Spark Sql 中的優化和 Hive 類似，可以參考Hive
0xFF 總結
數據傾斜的坑還是很大的，如何處理數據傾斜是一個長期的過程，希望本文的一些思路能提供幫助。
文中一些內容沒有細講，比如 Hive Sql 的優化，數據清洗中的各種坑，這些留待后面單獨的分享，會有很多的內容。
另外千億級別的數據還會有更多的難點，不僅僅是數據傾斜的問題，這一點在后面也會有專門的分享。
