廣播機制解決Shuffle過程數據傾斜的方法①

吳恩慈

(上海淇毓信息科技有限公司,上海 200120)

Spark 由加州大學伯克利分校AMP 實驗室開發,是基于RDD 的內存計算框架,在流式計算場景中表現良好.在Spark 分布式流計算集群的實踐過程中,往往出現數據分配不均勻的現象,某個分區的數據顯著多于其它分區,該節點的計算速度成為整個集群的性能瓶頸.任務執行慢的節點往往導致內存溢出,服務器CPU 使用率在短時間內急劇增加.

發生數據傾斜時,某個任務處理的數據量遠大于其他分片,從而增加了整個階段的完成時間.由于原始數據源的分布不均勻,每個Reducer 在分區映射過程中計算的數據量也不相同,任務執行時間不同,增加寬依賴階段的延遲,降低集群作業執行的效率.雖然支持用戶定義分區,但真正的數據分布難以預測,無法保證自定義分區功能的合理性和準確性.數據傾斜問題很難規避,為有效改善數據傾斜問題,本文在分析和研究國內外對該問題的研究成果和實踐經驗的基礎上,主要做了以下工作.

1)研究了Spark Shuffle 設計和算法實現,分析了哈希和排序兩類Shuffle 機制的實現過程,深入分析在Shuffle 過程發生數據傾斜的本質原因.

2)分析了Spark 流計算集群中,發生數據傾斜常見業務場景,以及數據傾斜問題的原因和發生過程,提供了問題定位的方法和步驟.

3)提出了Broadcast 機制避免某些場景下Shuffle導致的數據傾斜問題的方法,給出廣播變量分發機制和算法實現.通過Broadcast 實現Join 算子的實驗,驗證了該方法在性能上有較大提升.

1 相關工作

MapReduce 框架數據傾斜解決方案包括靜態和動態自適應調整.靜態自適應是事先分析數據集中鍵的分布特征,選擇適當的分區方法.該方法需要總結出一套和業務相關的數據集抽象規則算法,可以通過機器學習的方式來訓練算法,提高對各種類型數據集的適配度.文獻[1]提出的LEEN 方法,在Skew Reduce 的基礎上將分析過程移至Map 完成后,考慮Reduce 端輸入數據的公平性和數據本地性,LEEN 方法在緩解慢任務和Shuffle 引起的網絡擁塞方面有較大的性能提升[1].

動態自適應的思想是在應用運行時,實時檢測當前各個Partition 中的數據填充情況.如果發現存在嚴重的數據傾斜問題,在下一步進行調整,以避免原始MapReduce 任務之間的不能全局優化的問題.文獻[2]提出了Skew Tune 方法,推測當前任務需要完成時間,確定最可能的慢任務[2].掃描慢任務要處理的輸入數據,在數據偏斜場景中性能有較好的表現.文獻[3]提出在執行流程中插入一個Intermediate Reduce(IR)的新階段,用于并行處理動態的中間結果,使所有鍵值對都能夠利用到所有的節點資源[3].文獻[4]提出Adaptive MapReduce 策略,動態地調整每個Mapper 處理的輸入數據量,在Mapper 端使用一個固定大小的哈希表執行數據本地聚合[4].文獻[5]提出了任務重分割方法,通過監視任務的執行進度,根據預定策略直接切割任務本身[5].

本文與上述研究成果的不同之處在于通過Broadcast機制將數據分發到計算節點中,實現數據本地性,從根本上避免Shuffle 過程,不需要做額外的開發和部署.能夠快速解決特定場景下的數據傾斜問題,具有較高的實用性.

2 分區策略和Shuffle 算法

2.1 分區策略

分區策略(Partition)的主要作用是確定Shuffle 過程中Reducer 的數量,以及Mapper 側數據應該分配給哪個Reducer.分區程序可以間接確定RDD 中的分區數和分區中的內部數據記錄數.Spark 內置了Hash Partition 和Range Partition,支持自定義分區器.Hash Partition 的算法實現是獲取Key 的HashCode 除以子RDD 分區的數量取余.哈希分區器易于實現并且運行速度快.但有一明顯的缺點是不關心鍵值的分布情況,其散列到不同分區的概率會因數據而有差別.

Range Partition 在一定程度上避免了該問題.范圍分區器根據父RDD 的數據特征確定子RDD 分區的邊界,通過對數據進行采樣和排序,將父RDD 數據劃分為M個數據塊.如果數據均勻分布,則為每個分區提取的樣本大小都很接近.如果所有數據都分配給每個分區,則每個分區都會提取相同的數據量,并統計每個分區的實際數據量.若出現數據傾斜的情況,則對個別分區重新采樣.

2.2 Hash-Based Shuffle 算法

Spark 作為MapReduce 框架的一種實現,本質上將Mapper 端輸出的數據通過Partition 算法,確定發送到相應的Reducer,該過程涉及網絡傳輸和磁盤讀寫.Spark提供了哈希和排序兩類Shuffle 機制.ShuffleManager主要作用是提供Shuffle Writer 和Shuffle Reader 過程.

Hash-Based Shuffle 機制的實現過程包括聚合和計算,使用哈希表存儲所有聚合數據的處理結果.聚合和計算過程不進行排序,分區內部的數據是無序的,如果希望有序就需要調用排序操作.哈希Shuffle 強制要求在Map 端進行聚合操作,對于某些鍵值重復率不高的數據會影響效率.

如圖1所示,哈希Shuffle 算法實現過程.每個Mapper 都會根據Reducer 的數量創建一個相應的桶(Bucket).Mapper 生成的結果將根據設置的分區算法填充到每個桶中.當Reducer 啟動時,它將根據自己的TaskId 和它所依賴的MapperId,從遠程或本地Block Manager 獲取相應的Bucket 作為Reducer 的輸入.Bucket 是一種抽象概念,可以對應于某個文件或文件的一部分.哈希Shuffle 算法實現中,Mapper 會針對每個Reducer 生成一個數據文件,當Mapper 和Reducer數量比較多的時候,生成大量的磁盤文件.

圖1 Hash-Based Shuffle 算法實現過程

開啟consolidate 優化機制后,Shuffle Writer 過程中Task 不是為下游Stage 的每個Task 創建一個文件.每個ShuffleFileGroup 對應一批磁盤文件,磁盤文件數與下游Stage 中的Task 總數相同.Executor 分配多少個Core 就可以并行執行多少個任務,第一批并行執行中的每一個任務都創建一個ShuffleFileGroup,并將數據寫入相應的磁盤文件.當執行器執行下一批任務時,復用以前存在的ShuffleFileGroup 的磁盤文件,Task 將數據寫入現有磁盤文件.consolidate 機制允許不同的任務復用同一批磁盤文件,減少了文件數量并提高了Shuffle Writer 的性能.

2.3 Sort-Based Shuffle 算法

如圖2所示,Sort-Based Shuffle 算法實現過程.排序Shuffle 相比于哈希Shuffle,兩者的Shuffle Reader 過程是一致的,區別在Shuffle Writer 過程.排序Shuffle 允許Map 端不進行聚合操作,在不指定聚合操作的情況下,排序Shuffle 機制Mapper 端用數據緩存區（Buffer）存儲所有的數據.對于指定聚合操作的情況下,排序Shuffle 仍然使用哈希表存儲數據,聚合過程與哈希Shuffle 的基本一致.無論是Buffer 還是Hash Map,每更新一次都檢查是否需要將現有的數據溢存到磁盤中,需要對數據進行排序,存儲到一個文件中.更新完所有數據后,將多個文件合并為一個文件,并確保每個分區的內部數據都是有序的.

排序Shuffle 機制包括普通和bypass 兩種運行模式.當Shuffle Reader Task 的數量小于等于bypass Merge Threshold 的值時就會啟用bypass 模式.在普通操作模式中,首先將數據寫入內存數據結構,根據不同的Shuffle 運算符選擇不同的數據結構.如ReduceByKey 之類的聚合運算符選擇哈希數據結構,Join 類的普通運算符使用數組數據結構.在將每條數據寫入內存之后,確定是否已達到臨界閾值.如果達到臨界閾值則將內存中的數據寫到磁盤,然后清空內存中的數據.

在溢出到磁盤文件之前,根據Key 對內存中已有的數據進行排序,排序后數據將批量寫入磁盤文件.排序后的數據以每批10 000 個批量寫入磁盤文件,從而有效減少磁盤IO 數量.將Task 的所有數據寫入內存數據結構的過程中,會發生多次磁盤溢出操作,生成多個臨時文件,最后合并所有先前的臨時磁盤文件.由于一個Task 僅對應于一個磁盤文件,因此將單獨寫入索引文件以標識文件中每個下游Task 數據的開始和結束位置,磁盤文件合并過程減少了文件數量.

在bypass 模式下,為每個下游Task 創建一個臨時磁盤文件,并根據Key 的HashCode 寫入相應的磁盤文件.寫入磁盤文件時先寫入內存緩沖區,然后在緩沖區滿后溢出到磁盤文件.它還將所有臨時磁盤文件合并到一個文件中,并創建一個單獨的索引文件.此過程的磁盤寫入機制與未優化的哈希Shuffle 相同,但是Shuffle Reader 性能會更好.這種機制和普通的排序Shuffle 之間的區別是磁盤寫入機制不同,不會被排序.啟用bypass 機制的最大優點是在Shuffle Writer 過程中不需要執行數據排序操作,節省了部分性能開銷.

圖2 Sort-based Shuffle 算法實現

3 數據傾斜問題分析和定位

3.1 數據傾斜原因分析

發生數據傾斜的原因主要包括輸入數據源分布不均勻,以及計算過程中數據拉取時間不均勻.輸入數據分布不均勻的一般表現為原始數據,或者中間臨時數據中Key 分布不均勻[6].例如Spark Streaming 通過Direct Stream 讀取Kafka 數據.由于Kafka 的每個分區都對應于Spark 的相應任務,Kafka 相關主題的分區之間的數據是否平衡,直接決定Spark 在處理數據時是否會產生數據偏斜.如果使用隨機分區,概率上分區之間的數據是平衡的,不會生成數據傾斜.但很多業務場景要求將具備同一特征的數據順序消費,需要將具有相同特征的數據放于同一個分區中.一個典型的場景是相同的用戶相關的PV 信息放置在相同的分區中,很容易導致數據傾斜.

數據拉取時間不均勻一般是硬件計算能力不均勻,或者網絡傳輸能力不均勻造成的[7].比如PageRank 算子分為三個Stage 運行,由于第二個Stage 產生了Shuffle 是最容易發生數據傾斜,每個Task 處理分區數據綁定了各個頂點權重,然后收集其鄰接節點的權重.由于Executor 需要從非本地節點上拉取上一個Stage 中得到的節點信息.如果數據分布不均勻,某些節點會比其他節點承受更大的網絡流量和計算壓力.數據傾斜的計算時間主要花費在Shuffle 上,提高Shuffle 性能有利于提高應用程序的整體性能.

3.2 數據傾斜定位方法

通過Spark 界面觀察每個階段任務當前分配的數據量,進一步確定數據的不均勻分布是否導致了數據傾斜[8].只要在代碼中看到Shuffle 類操作符,或在Spark SQL 語句中看到導致Shuffle 的語句,就可以確定劃分Stage 的邊界.Stage 1 的每個任務開始運行時,將首先執行Shuffler Reader 操作,從Stage 0 的每個任務中提取需要處理的Key.比如Stage 1 在執行ReduceByKey 操作符之后計算出最終RDD,然后執行收集算子將所有數據拉到Driver 中.

數據傾斜往往發生在Shuffle 過程中,可能會觸發Shuffle 操作的算子包括GroupByKey、ReduceByKey和AggregateByKey 等.在執行Shuffle 時必須將每個節點上的相同Key 拖動到同一個節點上的Task 進行處理.如果某個Key 對應的數據量特別大就會發生數據傾斜,Job 運行得非常緩慢,甚至可能因為某個Task 處理的數據量過大導致內存溢出.

4 廣播機制避免Shuffle 的策略

4.1 Torrent Broadcast 原理

通過廣播機制將只讀變量從一個節點發送到其他Executor 節點,進程內運行的任務屬于同一個應用程序,在每個執行器節點上放置廣播變量可以由該節點的所有任務共享[9].Torrent Broadcast 算法的基本思想是將廣播變量劃分為多個數據塊.當某個執行器獲得數據塊時,當前執行器被視為數據服務器節點.隨著越來越多的執行器獲得數據塊,更多的數據服務器節點可用.廣播變量可以快速傳播到所有節點.Torrent Broadcast 讀取數據的方式與讀取緩存類似,使用Block Manager 自帶的NIO 通信方式傳遞數據,存在的問題是慢啟動和占內存[10].慢啟動指的是剛開始數據只在Driver 節點上,要等執行器獲取多輪數據塊后,數據服務節點才會變得可觀,后面的獲取速度才會變快.執行器在獲取完所有數據分塊后進行反序列化時,需要將近兩倍的內存消耗.

4.2 Broadcast 變量性能優勢

Driver 先把廣播變量序列化為字節數組,然后切割成BLOCK_SIZE 大小的數據塊.在數據分區和切割之后,數據分區元信息作為全局變量被存儲在Driver節點的Block Manager 中.之后每個數據分塊都做相同的操作,Block Manager Master 可以被Driver 和所有Executor 訪問到.執行器反序列化Task 時,先詢問所在的Block Manager 是會否包含廣播變量,若存在就直接從本地Block Manager 讀取數據.否則連接到Driver 節點的Block Manager Master 獲取數據塊的元信息.

廣播機制把只讀變量通過共享的方式有效的提高了集群的性能.大多數Spark 作業的性能主要消耗在Shuffle 環節,該環節包含了大量的磁盤IO、序列化、網絡數據傳輸等操作.通過廣播機制避免Shuffle,可顯著提高應用運行速度.例如普通Join 操作會產生Shuffle,RDD 中相同的Key 需要通過網絡拉取到同一個節點上.在算子函數使用廣播變量時,首先會判斷當前Task 所執行器內存中是否有變量副本.如果有則直接使用,如果沒有則從Driver 或者其他Executor 節點上遠程拉取一份放到本地執行器內存中.廣播變量保證了每個執行器內存中只駐留一份變量副本,Executor中的Task 執行時共享該變量副本,減少變量副本的數量和網絡傳輸的性能開銷,降低了執行器內存的開銷,降低GC 的頻率,會極大地提升集群性能.

4.3 Broadcast Join 代碼實現

如圖3所示,Broadcast Join 算子Scala 代碼實現.當在RDD 上使用Join 類操作或者在Spark SQL 中使用聯接語句時,普通聯接運算符會產生Shuffle 過程,并將相同的Key 數據拉入Shuffle Reader Task 進行聯接操作.如果連接操作中RDD 或表的數據量相對較小,則不使用連接運算符而是使用廣播變量和映射類運算符來實現Join 操作,從而完全避免了Shuffle 過程和數據歪斜.較小的RDD 中的數據通過收集操作直接拉入Driver 節點的內存中,創建廣播變量.然后從廣播變量中獲取較小RDD 的數據,并在另一個RDD 上執行映射類運算符.根據連接鍵比較當前RDD 的每個數據.如果連接鍵相同則將兩個RDD 的數據連接在一起,實現Join 操作的效果.該方法不足在于Driver 和Executor 節點都要存儲廣播變量的全部數據,比較消耗內存.

圖3 Broadcast Join 算子Scala 代碼實現

5 實驗和分析

5.1 集群硬件環境配置

本文使用的實驗環境是一個由四個節點組成的服務器集群.集群采用主-從體系結構,其中一個是主節點,其他三個是從節點.Spark Job 和Hadoop 文件系統部署在同一節點上.大部分的Spark 作業會從外部存儲系統讀取輸入數據,比如Hadoop 文件系統,將其與存儲系統放得越近越好.在相同的節點上安裝Spark Standalone 模式集群,并單獨配置Spark 和Hadoop 的內存和CPU 使用以避免干擾[11].實驗服務器集群環境的硬件配置根據數據量調整,參考官方建議初始配置如下.

單個節點服務器最初配置四個磁盤.Spark 很多計算都在內存中進行,但當數據在內存中裝不下的時候,仍然使用本地磁盤來存儲數據,以及在不同階段之間保留中間的輸出.在實驗中每個計算節點有4-8 個磁盤.集群網絡最初配置為萬兆網卡.當數據在內存中時,許多Spark 應用程序都與網絡密切相關.使用萬兆或更高的網絡是使這些應用程序更快的比較好的方法.分布式Reduce 應用程序尤其適用,比如group-by、reduce-by 等操作.

單個節點服務器的初始內存配置為16GB.Spark應用分配75%的內存,剩下的部分留給操作系統和緩沖區緩存.使用Spark 監視UI 的Storage 選項卡查看內存使用情況,內存使用情況受存儲等級和序列化格式的影響很大[12].單個計算節點的CPU 核心數量最初配置8 個Core.Spark 在每臺機器上可以擴展到數幾十個Core.測試階段在每臺機器上提供8-16 個內核,根據服務器負載的消耗可以配置更多.

5.2 系統配置和源數據

如圖4所示,集群運行時軟件版本信息,圖5是集群系統配置信息,都是當前業界使用的穩定版本和配置.實驗數據來自于某金融機構的客戶消費貸款逾期信息,包括逾期客戶還款信息,以及催收行動信息.選取逾期客戶還款信息數據量相對較少,10 萬條記錄,數據結構中包括記錄ID、還款時間、還款金額、銀行編碼、還款卡號、客戶姓名、客戶ID 等信息.逾期客戶機構催收行動數據量相對較大,1000 萬條記錄,數據結構包括記錄ID、催收行動碼、行動描述、客戶電話、客戶關系、通話時長、客戶ID、客戶姓名、行動時間、催收員、催收機構等信息.根據客戶ID 進行Join 操作,計算指標包括機構催收員工作量和催收員工作業績.

圖4 集群軟件版本信息

5.3 實驗結果分析

如圖6所示,Broadcast Join 算子DAG 視圖中不存在Shuffle,普通Join 算子DAG 視圖的復雜度明顯高于Broadcast Join 算子.圖6(a)是普通Join 操作的DAG 視圖,根據RDD 的寬依賴關系分為三個階段,有向無環圖描述了階段之間的依賴關系,當前Stage 只能在父Stage 之后執行.從DAG 視圖清晰的看到普通Join 算子存在Shuffle 過程.圖6(b)是Broadcast Join 算子有向無環圖只有一個階段,邏輯過程相對簡單.

圖5 集群系統配置信息

圖6 Join 算子DAG 視圖

如圖7所示,Broadcast Join 算子中Task 統計數據表明性能上存在明顯的優勢.圖7(a)是普通Join 算子某階段Task 性能統計數據,包括Task 持續時間、GC 執行時間和Shuffle 數據量等3 個方面的統計信息.圖7(b)是Broadcast Join 算子Task 性能統計數據,相比于普通Join 算子的Task 性能統計信息,在3 個方面都存在明顯的優勢.

如圖8所示,Broadcast Joins 算子各個Stage 的磁盤讀寫和網絡流量、任務持續時長等都存在明顯的優勢.圖8(a)是普通Join 算子各個階段執行情況,共有3 個階段,整個過程耗時6-7 秒,Shuffle 并行度為9,涉及Shuffle Reader 數據量61.3MB,Shuffle Writer 數據量61.3 MB.如圖8(b)所示,Broadcast Join 算子只有1 個階段,沒有涉及Shuffle 數據讀寫過程,數據輸入117.4 MB,Task 總數量明顯小于普通Join 算子的數量,執行時間是3 秒,和普通Join 算子相比在性能上有較大的提升.

圖7 Join 操作Task 性能數據

圖8 Join 算子各個Stage 執行情況

如圖9所示,Broadcast Joins 算子不存在數據傾斜問題.圖9(a)是普通Join 操作任務數據分配和執的詳細數據.從列Shuffle Read Size 可以看出,任務分配出現了數據傾斜問題,被分配數據量較大的Task 執行時間明顯高于其他任務的持續時間,消耗更大的資源和網絡流量,其他已經完成計算的節點處于等待狀態.圖9(b)是Broadcast join 算子任務數據分配和執行情況明細,從持續時間列和Input Size 列看出,數據幾乎是均勻分配,8 個任務的持續時間是1 秒,1 個是0.9 秒,充分發揮了數據本地性特性,每個節點的計算資源都被有效利用.

如圖10所示,Broadcast Join 算子在高并發的應用情況下性能上存在穩定的提升.如圖10(a)所示,普通Join 算子壓測統計,通過壓測工具10 萬次的統計結果,統計了平均持續時間、中位數和偏差情況.從圖中看出,持續時間絕大部分相對集中和穩定在7 秒左右.如圖10(b)所示,Broadcast join 算子壓測情況統計,比較相同的統計指標存在明顯的優勢,持續時間基本集中和穩定在4 秒左右.

6 總結

本文研究了Spark Shuffle 設計和算法實現,分析了哈希和排序兩類Shuffle 機制的實現過程,深入分析在Shuffle 過程發生數據傾斜的本質原因.進一步分析了Spark 流計算集群中,發生數據傾斜常見業務場景,分析數據傾斜問題的原因和發生過程,提供了問題定位的方法和步驟.提出了廣播機制避免某些場景下的數據傾斜問題,給出廣播變量分發機制和算法實現.通過Broadcast 實現Join 算子的實驗,相對于直接操作Join 算子,通過DAG 視圖、任務持續時間、Shuffle 讀寫數據量等指標的比較和分析,驗證了廣播機制在性能上有較大提升,壓力測試進一步驗證了在大規模應用的情況下性能有穩定的改善.

圖9 Join 算子數據分配和執行

圖10 Join 算子壓測數據統計