完全解密 Spark 的 HashShuffle

時間 2019-11-10

原文原文鏈接

本課主題

Shuffle 是分佈式系統的天敵
Spark HashShuffle介紹
Spark Consolidated HashShuffle介紹
Shuffle 是如何成爲 Spark 性能殺手
Shuffle 性能調優思考
Spark HashShuffle 源碼鑑賞

引言

Spark HashShuffle 是它之前的版本，如今1.6x 版本默應是 Sort-Based Shuffle，那爲何要講 HashShuffle 呢，由於有分佈式就必定會有 Shuffle，並且 HashShuffle 是 Spark之前的版本，亦便是 Sort-Based Shuffle 的前身，由於有 HashShuffle 的不足，纔會有後續的 Sorted-Based Shuffle，以及如今的 Tungsten-Sort Shuffle，因此咱們有必要去了解它。html

人們對Spark的印象每每是基於內存進行計算，但實際上來說，Spark能夠基於內存、也能夠基於磁盤或者是第三方的儲存空間進行計算，背後有兩層含意，第1、Spark框架的架構設計和設計模式上是傾向於在內存中計算數據的，第2、這也表達了人們對數據處理的一種美好的願望，就是但願計算數據的時候，數據就在內存中。算法

爲何再一次強調 Shuffle 是 Spark 的性能殺手啦，那不就是說，Spark中的「Shuffle「和「Spark徹底是基於內存計算「的願景是相違背的！！！但願這篇文章能爲讀者帶出如下的啓發：設計模式

瞭解爲何 Shuffle 是分佈式系統的天敵
瞭解 Spark HashShuffle的原理和機制
瞭解優化後 Spark Consolidated HashShuffle的原理和機制
瞭解Shuffle 是如何成爲 Spark 性能殺手
瞭解能夠從那幾方面思考 Spark Shuffle 的性能調優
瞭解 Spark HashShuffle 在讀、寫磁盤這個過程的源碼鑑賞

Shuffle 是分佈式系統的天敵

Spark 運行分紅兩部份，第一部份是 Driver Program，裏面的核心是 SparkContext，它驅動着一個程序的開始，負責指揮，另一部份是 Worker 節點上的 Task，它是實際運行任務的，當程序運行時，不間斷地由 Driver 與所在的進程進行交互，交互什麼，有幾點，第1、是讓你去幹什麼，第2、是具體告訴 Task 數據在那裏，例如說有三個 Stage，第二個 Task 要拿數據，它就會向 Driver 要數據，因此在整個工做的過程當中，Executor 中的 Task 會不斷地與 Driver 進行溝通，這是一個網絡傳輸的過程。緩存

[下圖是 Spark 官方網站上的經典Spark架框圖]
性能優化

在這個過程當中一方面是 Driver 跟 Executor 進行網絡傳輸，另外一方面是Task要從 Driver 抓取其餘上游的 Task 的數據結果，因此有這個過程當中就不斷的產生網絡結果。其中，下一個 Stage 向上一個 Stage 要數據這個過程，咱們就稱之爲 Shuffle。bash

思考點：上一個 Stage 爲何要向下一個 Stage 發數據？假設如今有一個程序，裏面有五個 Stage，我把它當作爲一個很大的 Stage，在分佈式系統中，數據分佈在不一樣的節點上，每個節點計算一部份數據，若是不對各個節點上獨立的部份進行匯聚的話，咱們是計算不到最終的結果。這就是由於咱們須要利用分佈式來發揮它自己並行計算的能力，然後續又須要計算各節點上最終的結果，因此須要把數據匯彙集中，這就會致使 Shuffle，這也是說爲何 Shuffle 是分佈式不可避免的命運。網絡

Spark 中的 HashShuffle介紹

原始的 HashShuffle 機制

基於 Mapper 和 Reducer 理解的基礎上，當 Reducer 去抓取數據時，它的 Key 究竟是怎麼分配的，核心思考點是：做爲上游數據是怎麼去分配給下游數據的。在這張圖中你能夠看到有4個 Task 在2個 Executors 上面，它們是並行運行的，Hash 自己有一套 Hash算法，能夠把數據的 Key 進行從新分類，每一個 Task 對數據進行分類而後把它們不一樣類別的數據先寫到本地磁盤，而後再通過網絡傳輸 Shuffle，把數據傳到下一個 Stage 進行匯聚。架構

下圖有3個 Reducer，從 Task 開始那邊各自把本身進行 Hash 計算，分類出3個不一樣的類別，每一個 Task 都分紅3種類別的數據，剛剛提過由於分佈式的關係，咱們想把不一樣的數據匯聚而後計算出最終的結果，因此下游的 Reducer 會在每一個 Task 中把屬於本身類別的數據收集過來，匯聚成一個同類別的大集合，抓過來的時候會首先放在內存中，但內存可能放不下，也有可能放在本地 (這也是一個調優勢。能夠參考上一章講過的一些調優參數)，每1個 Task 輸出3份本地文件，這裏有4個 Mapper Tasks，因此總共輸出了4個 Tasks x 3個分類文件 = 12個本地小文件。app

[下圖是 Spark 最原始的 Hash-Based Shuffle 概念圖]
負載均衡

HashShuffle 也有它的弱點：

Shuffle前在磁盤上會產生海量的小文件，此時會產生大量耗時低效的 IO 操做 (因為產生過多的小文件）
內存不夠用，因爲內存中須要保存海量文件操做句柄和臨時信息，若是數據處理的規模比較龐大的話，內存不可承受，會出現 OOM 等問題。

優化後的 HashShuffle 機制

在剛纔 HashShuffle 的基礎上思考該如何進行優化，這是優化後的實現：

[下圖是 Spark Consolidated Hash-Based Shuffle 概念圖]

這裏仍是有4個Tasks，數據類別仍是分紅3種類型，由於Hash算法會根據你的 Key 進行分類，在同一個進程中，不管是有多少過Task，都會把一樣的Key放在同一個Buffer裏，而後把Buffer中的數據寫入以Core數量爲單位的本地文件中，(一個Core只有一種類型的Key的數據)，每1個Task所在的進程中，分別寫入共同進程中的3份本地文件，這裏有4個Mapper Tasks，因此總共輸出是 2個Cores x 3個分類文件 = 6個本地小文件。Consoldiated Hash-Shuffle的優化有一個很大的好處就是假設如今有200個Mapper Tasks在同一個進程中，也只會產生3個本地小文件；若是用原始的 Hash-Based Shuffle 的話，200個Mapper Tasks 會各自產生3個本地小文件，在一個進程已經產生了600個本地小文件。3個對比600已是一個很大的差別了。

這個優化後的 HashShuffle 叫 ConsolidatedShuffle，在實際生產環境下能夠調如下參數：

 
         spark.shuffle.consolidateFiles= 
         true

Consolidated HashShuffle 也有它的弱點：

若是 Reducer 端的並行任務或者是數據分片過多的話則 Core * Reducer Task 依舊過大，也會產生不少小文件。

Shuffle是如何成爲Spark性能殺手及調優勢思考

Shuffle 不能夠避免是由於在分佈式系統中的基本點就是把一個很大的的任務/做業分紅一百份或者是一千份，這一百份和一千份文件在不一樣的機器上獨自完成各自不一樣的部份，咱們是針對整個做業要結果，因此在後面會進行匯聚，這個匯聚的過程的前一階段到後一階段以致網絡傳輸的過程就叫 Shuffle。在 Spark 中爲了完成 Shuffle 的過程會把真正的一個做業劃分爲不一樣的 Stage，這個Stage 的劃分是跟據依賴關係去決定的，Shuffle 是整個 Spark 中最消耗性能的一個地方。試試想一想若是沒有 Shuffle 的話，Spark能夠完成一個純內存式的操做。

 
         reduceByKey，它會把每一個 Key 對應的 Value 聚合成一個 value 而後生成新的 RDD

Shuffle 是如何破壞了純內存操做呢，由於在不一樣節點上咱們要進行數據傳輸，數據在經過網絡發送以前，要先存儲在內存中，內存達到必定的程度，它會寫到本地磁盤，(在之前 Spark 的版本它沒有Buffer 的限制，會不斷地寫入 Buffer 而後等內存滿了就寫入本地，如今的版本對 Buffer 多少設定了限制，以防止出現 OOM，減小了 IO)

Mapper 端會寫入內存 Buffer，這個便關乎到 GC 的問題，而後 Mapper端的 Block 要寫入本地，大量的磁盤與IO的操做和磁盤與網絡IO的操做，這就構成了分佈式的性能殺手。

若是要對最終計算結果進行排序的話，通常會都會進行 sortByKey，若是以最終結果來思考的話，你能夠認爲是產生了一個很大很大的 partition，你能夠用 reduceByKey 的時候指定它的並行度，例如你把 reduceByKey 的並行度變成爲1，新 RDD 的數據切片就變成1，排序通常都會在不少節點上，若是你把不少節點變成一個節點而後進行排序，有時候會取得更好的效果，由於數據就在一個節點上，技術層面來說就只須要在一個進程裏進行排序。

 
         能夠在調用 reduceByKey()接著調用 mapPartition( )； 
        
         也能夠用 repartitionAndSortWithPartitions( )；

還有一個很危險的地方就是數據傾斜，在咱們談的 Shuffle 機制中，不斷強調不一樣機器從Mapper端抓取數據並計算結果，但有沒有意會到數據可能會分佈不均衡，何時會致使數據傾斜，答案就是 Shuffle 時會導政數據分佈不均衡，也就是數據傾斜的問題。數據傾斜的問題會引伸不少其餘問題，好比，網絡帶寬、各重硬件故障、內存過分消耗、文件掉失。由於 Shuffle 的過程當中會產生大量的磁盤 IO、網絡 IO、以及壓縮、解壓縮、序列化和反序列化等等。

Shuffle 性能調優思考

Shuffle可能面臨的問題，運行 Task 的時候纔會產生 Shuffle (Shuffle 已經融化在 Spark 的算子中)

幾千臺或者是上萬臺的機器進行匯聚計算，數據量會很是大，網絡傳輸會很大
數據如何分類其實就是 partition，即如何 Partition、Hash 、Sort 、計算
負載均衡 (數據傾斜）
網絡傳輸效率，須要壓縮或解壓縮之間作出權衡，序列化和反序列化也是要考慮的問題

具體的 Task 進行計算的時候盡一切最大可能使得數據具有 Process Locality 的特性，退而求其次是增長數據分片，減小每一個 Task 處理的數據量，基於Shuffle 和數據傾斜所致使的一系列問題，能夠延伸出不少不一樣的調優勢，好比說：

Mapper端的 Buffer 應該設置爲多大呢？
Reducer端的 Buffer 應該設置爲多大呢？若是 Reducer 太少的話，這會限制了抓取多少數據
在數據傳輸的過程當中是否有壓縮以及該用什麼方式去壓縮，默應是用 snappy 的壓縮方式。
網絡傳輸失敗重試的次數，每次重試之間間隔多少時間。

Spark HashShuffle 源碼鑑賞

咱們說 Shuffle 的過程是Mapper和Reducer以及網絡傳輸構成的，Mapper 這一端會把本身的數據寫入本地磁盤，Reducer 這一端會經過網絡把數據抓取過來。Mapper 會先把數據緩存在內存中，在默應狀況下緩存空間是 32K，數據從內存到本地磁盤的一個過程就是寫數據的一個過程。

這裏有兩個 Stage，上一個 Stage 叫 ShuffleMapTask，下面的一個 Stage 多是 ShuffleMapTask，也有多是 ResultsTask，取決於它這個任務是否是最後一個Stage所產生的。ShuffleMapTask會把咱們處理的RDD的數據分紅苦幹個 Bucket，即一個又一個的 Buffer。一個Task怎麼去切分具體要看你的 partitioner，ShuffleMapTask確定是屬於具體的 Stage。

HashShuffle 寫數據的過程

在一個 Task 中，核心的代碼是 runTask，你能夠看到裏面建立了一個 ShuffleWriter，它是負責把緩存中的數據寫入本地磁盤的，但 ShuffleWriter 寫入入本地磁盤時，還有一個很是重要的工做，就是要先跟Spark 的Driver 通訊，說我把數據寫到了什麼地方，這樣下一個Stage找上一個Stage的數據的時候，它是找 Driver(blockManagerMaster)去獲取數據信息的，Driver(blockManagerMaster) 會告訴下一個Stage中的Task寫入的數據在那裏。
[下圖是 ShuffleMapTask.scala 中的 runTask 方法]
而後建立了一個 ShuffleManager，這是從 SparkEnv 中得到到的 ShuffleManager，SparkEnv是運行時的環境，因此在寫代碼的時候能夠配置它。
[下圖是 SparkContext.scala 中的 shuffleManager 變量]
再往下看ShuffleManager調用了getWriter 方法，在這裏咱們主要的是看 HashShuffle 的方式，因此看看它具體子類該怎麼實現。
[下圖是 ShuffleManager.scala 中的 ShuffleManager 類]

[下圖是 HashShuffleManager.scala 中的 getWriter 方法]
從getWriter方式建立了 HashShuffleWriter 的實例對象，因此若是須要看它具體的怎麼寫數據的話，必須要看 HashShuffleWriter 類，而後它也必需有一個 write 的方法，首先它會判斷一下是否有在 Mapper 端進行 aggregrate 的操做，也就是說是否進行的 Mapper 和 Reducer 這種計算模型的 LocalReduce，若是有的話，就基於records 進行聚合，它就會循環遍歷Buffer裏面的數據。在本地的聚合顯現帶來的好處是減小的磁盤IO的數據、以及操做磁盤IO的次數、以及網絡傳輸的數據量、以及這個 Reduce Task 抓取 Mapper Task 數據的次數，這個意義確定是很是重大的。
[下圖是 HashShuffleWriter.scala 中的 write 方法]
ShuffleWriterGroup，它會把相應的 Key 合併在同一個文件中，而後它會判斷一下是否須要進行一個合併的過程，它構建了一個ShuffleWriterGroup的實體對象，同時呢，它會判斷是否啓動壓縮機制，若是啓動了壓縮機制，會有一個fileGroup(bucketId)，不然的話就getFile(bucketId)。
[下圖是 ShuffleWriterGroup.scala 中的 forMapTask 方法]

[下圖是 ShuffleWriterGroup.scala 中的 forMapTask 方法內部具體的實現]
最後不管它採用那種狀況，最終也會調用 blockManager.getDiskWriter( )來完成寫入數據到本地磁盤這個動做
[下圖是 BlockManager.scala 中的 getDiskWriter 方法]
通過第4步後再回來看看shuffle，第一個參數是 shuffleId，第二個是 mapId，第三個是輸出的 Split 個數，第4個是序列化器，第五個是metric 來統計它的一些基本信息
[下圖是 HashShuffleWriter.scala 中的 shuffle 參數]
這裏的writer是代表具體要寫到什麼地方，bucketId 是經過傳入key到partition中的方式，下面調用 write 時有兩個參數，elem._1 和 elem._2，因此elem._1是key和elem._2是具體內容自己。當分好bucketId就開始寫數據。
[下圖是 HashShuffleWriter.scala 中的 runTask 方法內部具體的實現]

Spark 的並行度是繼承的，若是上游有4個並行任務的話，下游也會有4個
[下圖是 Partitioner.scala 中的 HashPartitioner 類以及它的方法實現]

[下圖是 Utils.scala 中的 nonNegativeMod 方法]

HashShuffle 讀數據的過程

在 Reader 中重點是看它的 Read 方法，首先會建立一個 ShuffleBlockFetcherIterator，這裏有一個很重要的調優的參數，也就是說一次能最大的抓取多少數據過來，在 Spark1.5.2 默應狀況下是 48M，若是你內存足夠大以及把內存空間分配足夠的狀況下，由於Shuffle會佔用百分比，能夠試試調大這個參數，調大這個參數的好處是減小抓取次數，由於網絡IO的開銷來創建新的鏈接其實很耗時的；往下看它再次進行一下判斷看看Mapper端的Aggregrator 是否已經定義了。
[下圖是 HashShuffleReader.scala 中的 read 方法]

[下圖是 HashShuffleReader.scala 中的 read 方法的內部實現]
在Shuffle寫數據的過程當中，一開始會建立 ShuffleBlockFetecherIterator 對象實例，而後調用它的 initialize( )方法
[下圖是 ShuffleBlockFetecherIterator.scala 中的 ShuffleBlockFetecherIterator 類]

[下圖是 ShuffleBlockFetecherIterator.scala 中的 localBlocks 和 remoteBlocks 變量]

[下圖是 ShuffleBlockFetecherIterator.scala 中的 initialize 方法]

在 initialize( )方法轉過來會調用 sendRequest( )方法，抓到數據後這裏有一個 BlockFetchingListener，它會對數據進行處理，
[下圖是 ShuffleBlockFetecherIterator.scala 中的 sendRequest 方法]
說明一點就是它底層有一套通訊框架，咱們基於這套通訊框架進行數據的請求和傳輸
[下圖是 NettyBlockTransferService.scala 中的 fetchBlocks 方法]

咱們從 Reducer端藉助了 HashShuffleReader 從遠程抓取數據，抓取數據過來以後進行 Aggregrate 操做匯聚，匯聚具體是進行分組或者是什麼樣的算法是開發者本身決定的。reduceByKey和Hadoop中的mapper與reducer相比，有一個缺點，在 Hadoop 的世界，不管你的數據的什麼樣的類型你均可以自定義，Mapper和Reducer的業務邏輯能夠完成不同。

Reducer端若是內存不夠寫磁盤的代價是雙倍的，在 Mapper端不管內存夠不夠它都須要先寫磁盤，由於Reducer端在計算的時候須要又一次的把數據從磁盤上抓回來，因此實際生產環境下須要適當地把 Shuffle 內存調大一點。

總結

由於想利用分佈式的計算能力，因此要把數據分散到不一樣節點上運行，上游階段數據是並行運行的，下游階段要進行匯聚，因此出現Shuffle，若是下游分紅三類，上游也須要每一個Task把數據分紅三類，雖然有可能有一類是沒有數據，這無所謂，只要在實際運行時按照這套規則就能夠了，這就是最原始的 Shuffle 過程。

Hash-based Shuffle 默認Mapper 階段會爲Reducer 階段的每個Task單首創建一個文件來保存該Task中要使用的數據，可是在一些狀況下(例如說數據量很是龐大的狀況) 會形成大量文件的隨機磁盤IO操做且會性成大量的Memory消耗(極易形成OOM)。