基于兩階分區的MapReduce實驗室系統負載均衡研究
2023-04-26 08:21:48鄭文麗熊貝貝程立勛蔡伊娜包先雨
計算機測量與控制 2023年4期
鄭文麗，熊貝貝，2，程立勛，蔡伊娜，2，包先雨
(1.深圳市檢驗檢疫科學研究院，廣東 深圳 518045；2.深圳海關食品檢驗檢疫技術中心，廣東 深圳 518000)
0 引 言
在數字化信息時代背景下，通過實驗室系統存儲和處理海量數據，已成為日常檢驗檢測工作不可或缺的一部分[1]。而隨著實驗室系統智慧化的要求，改進存儲和處理技術在大規模數據高效采集領域中越發重要。其中，MapReduce是一種處理大規模數據集的分布式框架，擁有拓展性強和容錯率高的特點。通過Mapper和Reducer對海量的檢測數據進行拆分和歸類，與傳統構架相比，能更高效地處理數據。而Hadoop框架源于Apache Nutch，集合了HDFS(hadoop distributed file system)，YARN(yet another resource negotiator)和MapReduce，是具備高容錯性、高拓展性和低成本特點的分布式軟件框架，廣泛應用于各種領域，包括分布式搜索[2-3]、模式識別[4-5]、計算機機器學習[6-7]和文獻聚類[8-9]等。
MapReduce框架主要運用于大型云平臺的海量數據庫存儲的應用場景中，可解決數據采集高頻、數據類型多樣、數據來源復雜以及用戶需求響應及時等難點痛點。同時，根據不同場景的業務應用流程，云平臺的設計與開發通常采用分層式管理并按需優化，這為大數據的分流工作提出了新的挑戰。王立俊等人[10]在設計氣象大數據云平臺時，其總體架構共分為基礎設施層、數據管理與處理層和應用層，實現了更快的數據采集以及更優的數據處理功能。鑒于氣象數據屬于典型的時序性數據，其數據流量高達6萬次/秒，利用MapReduce這種高性能的分布式計算框架則可以大幅簡化數據處理流程。在該云平臺中，其MapReduce任務包括了3個任務階段，分別為Map階段、Shuffle階段以及Reduce階段，這3個階段可連接為鏈式工作流。數據流在Map階段，依據數據的不同屬性和特點進行處理之后，形成不同長度的數據塊，以相應的規則映射成鍵值對，這些鍵值對就是實例化的Map任務。經過Map階段處理后的分片數據，生成
MapReduce除了在云平臺系統運用方面有很多理論研究基礎，在其框架的改良優化方面也有著廣泛的研究成果，胡東明等人[11]在MapReduce框架下提出了一種負載均衡的Top-k連接查詢算法。該算法不僅在MapReduce框架下實現了Top-k連接查詢算法，還通過提前終止機制和負載均衡機制來增強其數據連接處理性能。由于Map任務和Reduce任務可以并行處理的特點，能盡量避免了鏈接MapReduce作業的初始化開銷，降低了數據處理成本。該算法流程包括直方圖構建、提前終止機制、數據過濾、負載均衡機制4個步驟組成，其中提前終止、數據過濾都在Map階段實現，Reduce階段則通過Top-k連接查詢算法完成數據清洗。在數據過濾階段，Map任務會處理每個記錄，因為針對每個作業會形成不同的過濾機制。通過了數據過濾機制的記錄則進入到Reduce階段，并使用啟發式任務調度算法對每個記錄的Reduce任務進行數據調度，將這些任務依次分配給連接總數較低的Reducer進行處理。該算法利用了Map階段輸出的鍵值對以及連接值進行分組，將其按照自定義分區程序的結果分配給Reduce任務。MapReduce的并行化處理大大降低了算法的總執行時間，但采取不同Top-k算法則產生不同的任務執行時間，經比較，最終選用了P-TKJ算法，相較于RSJ算法獲得了更快的執行速度，并隨著數據集的增大，其效率的優勢越明顯，使得MapReduce框架下不僅實現了海量數據的Top-k連接查詢算法，還提高了CPU的利用效率。
在實驗室系統中引入MapReduce框架[12]，可以極大簡化分布式程序，集中精力于數據處理的任務本身，提高實驗室系統數據處理的效率[13]。實驗室系統在執行MapReduce任務時，數據是以鍵值對輸入，并在Mapper節點時進行聚合處理，最終根據哈希值函數Shuffle至各Reducer節點進行Sort和Reduce。在風力發電行業中，其實驗室監測數據包含設備的狀態參數、氣象環境、地理信息等，數據體量較大，類型多。雖然Hadoop平臺能夠滿足大數據的基本需求，但在運營效率方面仍有待提高。王林童等人[14]提出了基于MapReduce的多源數據并行關聯查詢的優化方法，主要是針對風電大數據進行存儲預處理，盡量使具有關聯關系的數據存儲在同一存儲節點中，之后采用哈希分桶算法對數據存儲進行優化，并在查詢時利用MapReduce框架的并行性特點，最終對數據可以采用并行優化的查詢與計算。具體而言，該系統首先對具有相同屬性(時間、地點、設備號等)的風電數據進行歸并的預處理，在關聯查詢的基礎上數據清洗之后得到風電時序數據合并表。合并表按照關聯字段的哈希值分配到不同的“桶”中，同一個“桶”中的數據即存儲到同一個數據節點中，實現本地化的存儲優化。當用戶使用該系統進行數據查詢時，設計在Map階段完成風電數據的查詢、過濾、篩選等操作，盡量減少Reduce階段的操作，此時就大大降低了傳輸時延，提升了數據查詢效率。文獻[15]則利用MapReduce框架，構建了一個基于物聯網的細胞生物學智慧云實驗室系統，該模型共包括四層：實驗設備層、IT基礎設施層、控制層以及數據分析層。實驗設備層主要工作為自動采集不同實驗儀器設備上產生的數據，采集后的實驗數據交由IT基礎設施層進行存儲，在控制層可對這些存儲的數據進行查詢，而在數據分析層，則針對數據種類應用不同的分析方式進行篩選；通過4個層級的相互作用，為實時根據用戶的需求進行針對性調整和優化提供了極大的便利。
在這個過程中，如果原始數據分布不均，易出現數據傾斜的問題，引發Reducer節點負載不均衡，導致整個MapReduce任務的執行時間過長。因此，基于MapReduce的實驗室系統負載均衡成為近期國內外學者們的研究熱點。
1 基于兩階分區的MapReduce負載均衡算法
針對原始數據分布不均導致的數據傾斜問題，國內外學者從不同角度提出了解決方案。杜鵑等人提出了一種利用快速無偏分層圖抽樣算法的負載均衡算法，在小規模數據上運行良好，但未在大型真實的復雜數據集中運用、驗證[16]。陶永才等人提出了MR-LSP(MapReduceon-line load balancing mechanismbasedon sample partition)算法，對原始數據進行采樣分析，通過分析結果進行負載均衡的分區分配Reducer節點的策略，但該算法忽略了數據采樣率不高時的情況[17]。在馬青山等人[18]提出的DSJA(data skew join algorithm)算法從數據關系表中的連接鍵出現頻率的角度出發，區分數據是否產生連接傾斜的情況，再分配到相應的Reducer節點中進行處理，但該算法僅考慮了Reduce端輸出的負載均衡，未考慮Map端到Reduce端的輸入階段的負載均衡處理。M.A.Irandoost 等人提出的LAHP(learning automata hash partitioner)算法[19]是根據學習自動機策略，在作業執行階段，對數據鍵值進行調配各個Reducer節點的數據量；但該算法只考慮了數據偏度高，而未考慮數據采樣率高的情形，且只優化了執行階段，忽略了計算時間對流程的影響。Elaheh Gavagsaz 等[20]提出的可拓展的高偏度數據的隨機采樣模型SBaSC(sorted-balance algorithm using scalable simple random sampling )對高偏度數據進行了隨機采樣，通過采樣中鍵值近似分布推算數據的分布，從而進行負載平衡的分配；該算法在基于Spark的網絡數據中得到了較好的效果，但未能泛化至不同的數據集，存在一定的局限性。黃偉建[21]等使用并行隨機抽樣貪心算法，縮短了采樣階段的執行時間，但當MapReduce輸入數據量較大時，準確性不夠且負載效果較差。
為解決上述問題，Xu等[22]提出的兩階分區算法在數據偏度較低時通過CC(cluster combination)調度方案實現負載平衡，在數據偏度較高時則通過CSC(cluster split combination)調度方案，可比傳統的MapReduce減少60%的運算時長。在該算法中，一個數據處理作業被分為了兩個階段，一是制定分區方案，二是執行MapReduce作業。在制定分區方案中，CC調度方案主要思路是選擇具有最大數量的鍵值對的數據塊調度給當前最小工作量的鍵值對的Reducer，這種啟發式算法使用了標準差這一評價指標來自適應地衡量所有Reducer的負載量。當標準差高于設定的閾值后，將對數據塊進行再分配。其分配的方式基于對數據塊所含鍵值對數量的降序排序，并對Reducer也進行工作量進行降序排列，接下來則將兩個表中的首位進行匹配，以確保整個MapReduce系統的負載均衡。CC調度方案可以解決數據量輕度傾斜的情況，但面對數據量極度傾斜的情況，就無法實現較好的效果。針對這樣的情況，該算法又提出了第二種情況的方法，即CSC調度方案。面對包含著較多數據記錄的數據塊，CSC采用的是劃分的方式，將較大的數據塊劃分成較小的數據塊。由于在MapReduce框架下，每個數據塊都對應著一個Reducer，因此在對較大數據進行劃分的時候，需要額外分配一個Reducer。與CC類似，CSC在進行數據塊分割的時候也是采用啟發式算法解決這個NP-Hard(non-deterministic polynomial hard)問題，偏度較大的數據塊將被分成n塊(n為Reducer的總數)，以確保整個MapReduce系統保持負載平衡。在分區任務完成之后，該算法便執行MapReduce任務。在MapReduce任務階段主要是對數據集進行采樣，不同的數據則有不同的特點，因此，不同的數據集則采用不同的采樣方法，如對預訓練過的數據集，則采用區間分布；對于全新的數據集，則采用隨機抽樣的方式。該算法為了具備更強的泛化性能，采取了隨機抽樣的方式，并利用Map任務和Reduce任務的并行處理特點，先對數據集進行預處理，利用鍵值中位數估值的概率進行分布。首先對數據集進行均勻的數據采樣，再針對采樣中出現鍵值的概率分布進行統計，基于統計結果，擬合不同的概率分布，并根據相應概率分布對數據塊進行抽樣，以解決數據偏度和采樣率都較高的問題。雖然該兩階分區算法在解決數據偏度和采樣率都較高的問題中都有相當好的性能表現，但在執行MapReduce作業，該算法也存在著以下的缺陷：
1)由于兩次MapReduce任務采用了串行的執行方式，導致整體任務的執行時間長。
2)在采樣率較高時，由于抽樣數據會在執行階段進行重復的MapReduce作業，處理時間過長。
3)執行CSC調度方案時，資源利用率較低，導致出現了部分Reducer節點出現空轉的情況。
為了解決上述算法執行時間較長及負載不均衡的問題，本文提出了一種基于MapReduce框架下兩階分區的改進算法。該算法主要工作為：
1)對MapReduce任務的流程進行并行化處理。
2)在采樣率高的情況下，將采樣階段輸出數據進行回收。
3)在原始數據偏度較高的情況下，提出ICSC(improved cluster split combination)算法。本文提出的算法可以在算法速率提高且執行時間縮短的同時，有針對性地解決高偏度，高采樣率情況下數據傾斜的問題。改進后兩階段的流程如圖1和圖2所示。
圖1 采樣階段流程
圖2 執行階段流程
在采樣階段，與優化前的兩階分區算法不同的是，之前是先進行采樣數據進行數據統計再劃分數據塊，優化后則是先對數據塊進行劃分，并對劃分后的數據塊進行采樣后與數據集的概率分布進行比對，當某一個數據塊的采樣率很高(超過設定的閾值)時，就不需要再重新對數據塊進行采樣而直接使用采樣出的數據。低于采樣率閾值的數據塊則需要再進一步采樣，但也是增加而非重新分配。這樣的做法不僅能夠降低時間開銷，同時也降低了系統對于數據取用的I/O成本。
而在MapReduce執行階段，本文算法除了使用其并行處理的特點，在數據被劃分為不同數據塊后被分配給不同的Mapper，由于數據塊在采樣階段就出現了處理速度不一致的情況，因此如果采用串行方式則會大大拖累不需要重采樣的數據塊，增加時間開銷。本文算法通過直接處理和計算系統負載均衡的方式并行處理數據塊，并在分配好數據塊的Reducer之后再進行不同數據偏度方式進行調度。對于偏度較低的數據塊，本文依然沿用了兩階分區算法中的CC調度方式，而針對偏度較高的數據塊則采用了優化了CSC調度方式的ICSC方法，具體的過程在第2節中進行詳細地闡述。
2 算法設計
2.1 兩階并行的設計
根據Apache官方文件所述，當系統批處理數據且執行MapReduce任務時，Mapper和Reducer的任務可以進行一定程度的并行處理。其中，Mapper需完成Map的子任務，并將輸入的鍵值對進行歸類。而Reducer需要完成的任務主要分為3個子階段：Shuffle、Sort和Reduce。Shuffle階段的職責為根據提供的哈希函數將Map任務的結果分別傳輸到各個Reducer節點上，同時也是唯一一個可以和Map任務同時進行的子任務。因此，Shuffle子任務的結束時間至少晚于Map任務的結束時間。Sort階段是各個Reducer將收到的數據進行排序、統合的過程。最后的子階段Reduce則是對數據進行計算，并將結果輸出到HDFS的過程。在Hadoop的MapReduce框架下，可根據Map的任務進度，通過調整Slowstart參數，對Shuffle的開始時間進行自定義。這種若參數設置為1.00，則為串聯運行，Shuffle子任務將在Map任務全部完成后再執行。
在兩階分區的算法中，由于執行階段Shuffle子任務并不依賴于Map任務的完成，而是依賴于上一個階段——采樣階段的Reducer結果所產生的分區方案，因此采樣階段和執行階段并非兩個相互獨立的任務，所以沒有利用調整Slowstart參數來進行并行化。但在MapReduce任務中，Map任務的執行時間往往遠長于Shuffle任務的時間，所以若執行階段的Map和Shuffle并行處理，會導致Shuffle任務的執行時間被拉長。基于此，本文提出在第一階段得到采樣結果后便開始第二階段的Shuffle任務，通過并行化流程來減少Mapper節點的空轉導致的資源浪費。具體流程如圖3所示，實線箭頭標識了串行工作時段，虛線箭頭標示了并行工作時段。本文算法將原有的流程做兩方面更改：首先，讓執行階段MapReduce的Map任務隨著采樣階段Map任務的結束立刻開始；其次，讓執行階段MapReduce的Shuffle任務隨著采樣階段的分區方案的計算完成而立即啟動，通過并行化的方式縮短整體任務的執行時間。
圖3 MapReduce流程優化前后對比
2.2 采樣階段的設計
在兩階分區算法中，第一次MapReduce任務的結果只用來規劃執行階段各Reducer節點的負載分配，在數據計算完畢后分區處理后則被舍棄。該算法的設計在數據采樣率較高的時候，由于對同樣的數據進行相同的多次操作，就會造成計算資源嚴重浪費，增加數據使用開銷。針對這樣的問題，本文對上述算法進行了以下改進：將第一階段數據塊采樣輸出的結果重新Shuffle至對應的Reducer節點上。因此，到了執行階段的MapReduce，則可直接使用采樣數據塊，而不必對同一數據塊進行重復作業，使采樣數據得到充分利用，且節省了存儲資源與計算資源。
2.3 分區調度階段的設計
根據數據偏度情況，本文采用了兩種不同的分區調度算法。在偏度較低時，則繼續沿用Xu等人[22]提出的CC(cluster combination)調度算法，對采樣結果進行分區。在偏度較高時，在執行CC(cluster combination)調度算法，大數據塊會導致負載不均衡。因此，上述文獻中提出一種CSC(cluster split combination)調度算法來執行分區。該調度算法先對大數據塊進行拆分，由多個Reducer節點分別處理，當所有Reducer完成各自的任務后，還需要一個Reducer進行額外的合并任務。此時，其它Reducer會陷入空轉的狀態；同時，如果所有的Reducer都參與到最后一次Reduce任務中，任何一個節點的故障都會導致最后一次Reduce任務無法進行，致使MapReduce任務執行失敗，如圖4所示。
為此，本文提出的ICSC算法，如圖5所示。其主要思想是將Reducer節點的工作量進行一定程度的壓縮，使得這些Reducer完成工作的總時間與其他不參與處理大數據分片的Reducer節點保持一致，減少節點的空轉時間，更好地利用計算資源，也達到了數據并行處理使得系統優化的效果。同時，由于執行時間對齊，該算法產生的結果并不會因為任何一個節點的故障而執行失敗，意外產生的情況能夠獨立執行作業進行處理，提升了系統的穩定性。為實現以上功能，ICSC算法的偽代碼如下：
ICSC(Clusters C, Reducers R){
average = C.size/R.count;
C.sort();
largeC = new Array();
for (i = 1; i <= C.size; i++){
if (C[i - 1].size >= average){
largeC.add(C[i - 1]);
C.remove(i - 1);
}
}
// n為參與到包含split的reduce任務的節點數量。
n = ceiling(largeC.size/average)
for (i = 1; i <= n; i++){
R[i - 1].assign(largeC.split(n));
}
R.sort();
while(!C.empty){
C.assign(R);
R.sort();
}
}
算法1 ICSC算法偽代碼
圖4 CSC調度
圖5 ICSC調度
3 試驗與分析
3.1 試驗驗證
為模擬實驗室系統的數據收集和處理過程，本試驗采用WordCount算法和人工生成的數據，在同構環境下以固定大小的輸入數據進行實驗驗證本文提出的算法。由于實驗室系統的實際應用環境下，數據屬性中的偏度和采樣度是影響數據處理階段的主要因素，因此本文通過控制變量法，驗證在不同數據偏度和采樣度的情況下，ICSC算法可在實驗室系統的數據處理階段有效減少耗時，達到流程優化的效果。
具體而言，試驗以JAVA JDK 11.0.13編制程序模擬12節點的集群，在同構環境下使用WordCount算法處理128 MB人工數據。在實驗室系統的環境中，節點間的傳輸采用100 Mbps帶寬。測量時間使用currentTimeMillis函數來測量自采樣階段Mapper到Reducer完成最后一個任務為止的時間差值。 通過試驗得出結果如表1和表2所示。
表1 數據偏度在不同方式下對耗時的影響
表2 采樣度在不同方式下對耗時的影響
3.2 比較分析
為了更好地比較ICSC算法與CC和CSC調度法的耗時效果。根據在上述結果(表1和表2)的基礎上分別以輸入數據偏度和采樣率為自變量，以各自的執行時間為因變量，得出不同算法條件下，各自的偏度和采樣度的效率，如圖6和圖7所示。
圖6 偏度與時間對比
圖7 采樣度與時間對比
如圖6所見，當數據偏度大于0.1時，ICSC算法采樣時間更少，且偏度愈大優勢更明顯，通過測算，在處理偏度為1.5的數據時，對比CSC算法，ICSC算法可節約10%的執行時間；但當數據偏度小于0.1時，由于ICSC在最后需要進行一次額外的Reduce以合并被拆分的大數據塊，因此執行時間是會稍大于CC和CSC調度法所耗時間。
同樣在圖7中，可以看到在不同采樣度的環境下，ICSC與兩階分區中算法的執行時間對比。3種算法均在采樣度為0.1左右時執行效率最高；而在0.1及以上時，ICSC算法比其它兩種算法執行時間更短。這是因為在本地性較高的環境下，ICSC通過節省第二輪MapReduce的數據處理量以達到更優的執行時間。
4 結束語
本文針對兩階分區算法中在實驗室系統的實際數據處理場景中存在的不足，對其調度算法和執行流程進行改進。實驗結果證明，改進后的算法在數據存在高偏度和高采樣度的情況下，均可有效地減少了Mapper和Reducer節點空轉的時間，弱化了Reducer節點間的依賴性，縮短MapReduce的執行時間，優化了實驗室系統數據處理階段的流程，從而高效地實現MapReduce框架下的實驗室系統中數據處理的負載均衡。但值得注意的是該算法在本地性較差的環境下仍有待改進，即數據采集系統與數據處理系統之間需要一定時間開銷的情況，因為在這樣條件的環境下，數據處理節點間的傳輸時間遠大于MapReduce作業的計算時間，此時節點間的傳輸時間成為ICSC算法執行時間的主要影響因素。而ICSC算法主要針對計算時間進行優化，在執行時間的優化效果較為遜色，這也是本研究下一步工作努力的目標和方向。
