MapReduce的系統性能評估與Backup調度策略(一)

 摘要
 MapReduce是一個在海量數據上進行數據處理的并行編程模型，它特別適合于海量非結構化和結構化數據的搜索、分析和挖掘任務，已經開始被人們廣泛使用。對于興起的眾多類似MapReduce系統來說，如何有效地評估和分析對比這些系統，成為當前一個需要解決的問題。
 本文詳細討論了針對MapReduce運行系統的性能評估指標和方法，設計和選擇一系列具有代表性的程序和數據作為基準，用來評估和分析MapReduce系統。在這一評估方法指導下，本文在我們自己實現的MapReduce運行系統——Tplatform平臺上擴展了Profiling功能，然后進行了一系列評估實驗，來分析和尋找系統性能瓶頸，為未來系統優化提供依據。通過實驗我們發現了我們系統的一些可改進的問題如任務調度、落后者問題等等。我們選擇了針對導致提交任務延遲增加的落后者問題，通過實現后備任務策略來嘗試改進。經模擬實驗結果顯示，我們提出的改進策略能夠有效地改進落后者問題的性能問題。
 
 關鍵詞：MapReduce，性能評估，落后者問題，后備任務策略
 
 Abstract
 MapReduce is becoming an important parallel programming paradigm for processing Internet scale data. It is widely used to process jobs such as searching, analyzing, and mining on large scale structured and semi-structured data. It is still a problem for the emerging MapReduce-like systems to analyze and eva luate systematically and efficiently.
 This paper discussed the issues in performance eva luation for MapReduce runtime system. We designed and chose a series of representative programs and data as benchmark. And then we implement profiling in our homemade MapReduce system which named Tplatform. We did the eva luation experiment for finding the bottleneck of the system. Through the experiment, we found some performance problems such as scheduling and stragglers etc. We implemented backup tasks for improving the problems caused by stragglers. Our simulation results reveal that we improve the performance efficiently.
 
 Keywords: MapReduce, Performance eva luation, Stragglers, Backup tasks
 目錄
 第 1 章 引言 4
 第 2 章 MapReduce框架 6
 2.1 MapReduce模型介紹 6
 2.2 系統實現 6
 2.3 Tplatform的實現 8
 第 3 章 系統評估 10
 3.1 評估目標 10
 3.2 基準程序和數據 10
 3.2.1 基準程序集合 11
 3.2.2 評估目標 13
 第 4 章 系統監控和程序概要分析 15
 4.1 實現細節 15
 第 5 章 評估實驗 17
 5.1 機群配置 17
 5.2 實驗結果 17
 5.2.1 單任務延遲和總機器時間 17
 5.2.2 平均結束時間 18
 5.2.3 加速比 18
 5.2.4 公平性 20
 5.2.5 故障恢復穩定性 20
 5.3 實驗結果和性能問題分析 20
 5.4 開銷分析 22
 第 6 章 后備任務調度策略 24
 6.1 問題描述 24
 6.2 相關工作 24
 6.2.1 MapReduce 24
 6.2.2 Hadoop 25
 6.2.3 異構環境中后備任務調度 25
 6.3 實現細節 26
 6.3.1 整體框架 26
 6.3.2 落后者判定策略 26
 6.3.3 系統處理過程 28
 6.3.4 數據結構細節 28
 6.4 后備任務策略評估實驗 29
 6.4.1 機群配置和任務準備 29
 6.4.2 任務耗時趨同性分析 29
 6.4.3 后備任務策略評估 30
 第 7 章 系統優化方向 33
 7.1 網絡傳輸問題 33
 7.2 增加用戶和系統的交互 33
 7.3 從數據庫領域看系統性能的其他提升空間 34
 7.4 系統易用性 34
 第 8 章 總結 35
 
 
 
 
 引言
 MapReduce正在成為人們在海量數據上進行并行計算的重要編程模型，比如為大規模的網頁做索引、在海量的數據中進行挖掘、龐大的科學計算任務等等。
 人們開始關注在普通計算機上實現大規模的并行計算以提供各種服務，Google則無疑是這方面的先驅者。Google使用MapReduce作為日常計算的引擎，將每天處理20PB的數據[  Dean, J. and Ghemawat, S. MapReduce: Simplified Data Processing on Large Clusters. proceedings of OSDI, 2004.]存在底層的存儲系統如GFS6、BigTable7中。很多重要的搜索引擎服務，如索引、網頁排序、網頁消重與去噪、用戶日志分析、用戶行為預測等等，都可以使用MapReduce的框架來加快程序員在進行相關的處理。
 此外，MapReduce也是一個如今很受歡迎的并行計算模型。MapReduce良好的可擴展性使得并行處理變得很容易，人們可以很方便地把MapReduce部署到大規模的廉價機群上使用。它的開源實現版本Hadoop[  Hadoop. The hadoop project. http://lucene.apache.org/hadoop/, 2006.]也得到了廣泛的應用。如今很多公司如Yahoo!、FaceBook、Amazon、New York Times，以及部分研究機構和大學如CMU、Cornell等等都開始使用Hadoop進行研究和開發。
 為了更好和方便地讓程序員使用MapReduce或者類似的并行處理計算框架如Map/Reduce/Merge6，人們在其上架設了一系列的編譯系統，并通過高層的語言把計算任務映射為底層的MapReduce任務。這方面的工作如Yahoo! 在Hadoop上實現的Pig[  C. Olston, B. Reed, U. Srivastava, R. Kumar and A. Tomkins. Pig Latin: A Not-So-Foreign Language for Data Processing. proceedings of SIGMOD, 2008.]、Google實現的Sawzall[  Pike, R., Dorward, S., Griesemer, R., and Quinlan, S. Interpreting the data: Parallel analysis with sawzall. Scientific Programming 13, 4 (2005), 277–298.]等等。
 類似系統的開發和研究也層出不窮，如微軟有自己的Dryad5/SCOPE7/DryadLINQ[  Isard, M., Budiu, M., Yu, Y., Birrell, A., and Fetterly, D. Dryad: distributed data-parallel programs from sequential building blocks. proceedings of the ACM SIGOPS, 2007]系列系統。擁有這樣的處理能力無疑成為一個互聯網公司的核心競爭力，可以預見在未來的一段時間里面，還有類似的很多系統和研究出現。
 人們在使用Hadoop或者類似的其他并行處理計算框架及其上層語言時，眾多的使用者對底層大規模并行處理計算框架有自己的需求。比如大學或研究機構使用此類框架進行科學計算時，系統的工作負載可能是偏向計算密集型，人們也關心系統對于計算任務的延遲反應；而大型因特網公司如Google、Yahoo!、Microsoft Live Search等的數據中心中，有若干程序員在同時提交計算任務，程序員不但關心計算任務的延遲，還關心整個中心中負載的調度公平性；而對于此類系統的開發和研究人員來說，他們關心系統的吞吐量、系統中各機器的狀態和使用情況等等。所以考慮此類并行處理計算框架特別是MapReduce系統的各項系統指標，并確定評估的程序和方法，對評估類似系統、基于用戶希望的系統設計折衷進行系統之間的比較、改進系統等等有很重要的意義。在這個基礎上如Berkeley也有一些系統測試的工作如分析網絡的性能X-Trace[  R Fonseca, G Porter, RH Katz, S Shenker, I Stoica. X-trace: A pervasive network tracing framework. proceedings of NSDI , 2007.]，以及對MapReduce系統和數據庫系統性能評估的討論15。
 我們基于MapReduce實現了自己的并行處理計算框架，并在其之上進行了系統的測試和評估。我們提出了測試程序和數據，并基于此在系統中實現了監控和程序性能概要分析框架。通過測試和評估實驗，我們總結了系統的性能指標和觀察到的問題。我們針對其中的單機落后問題，實現并驗證了后備任務策略，并基于此改進系統性能。最后，我們總結并給出了其他工作方向。
 論文的剩余部分按如下方式進行組織。第二章對MapReduce的模型和體系結構進行概述，而第三章列出了需要評估的系統目標和我們設計的基準程序和數據集合。為了分析和評估系統，我們在第四章闡述了系統監控框架和程序概要分析的設計和實現細節。之后我們在第五章中列出了實驗結果和給出了實驗的分析，并在針對其中的落后者問題實現了后備任務策略，在第六章中詳細闡述了后備任務策略的實現和實驗評估。我們在第七章中對系統可能的優化方向進行了展望并在第八章中進行了總結，最后是致謝。

 MapReduce框架
 在這一章里面，我們將簡單介紹MapReduce框架的模型和我們的系統實現。
MapReduce模型介紹
 Google的研究人員受到函數式編程語言（functional language）的啟發，在總結大量的大規模分布式處理程序共同特征的基礎上，提出了MapReduce并行程序框架。
 MapReduce是一大類大規模并行數據處理程序的抽象。這類計算的輸入是一個（鍵，值）對的集合，輸出也是一個（鍵，值）對的集合。用戶只需要提供兩個操作map和reduce的實現，MapReduce運行時庫就可以自動把用戶程序并行化。
 用戶提供Map函數的實現，它接收一個輸入對，產生一組中間結果對。MapReduce庫會把具有相同鍵的所有中間結果對聚合到一起，把他們傳給Reduce函數。用戶提供的Reduce函數，接收中間結果的一個鍵和具有此鍵的一組值，處理這些值，產生若干個（鍵，值）對做為輸出。它們的一般形式如下：[  楊志豐，GFS與MapReduce的實現研究及其應用，北京大學碩士論文，2008.]
 Map (k1, v1) -> list (k2, v2)
 Reduce (k2, list (v2)) -> list (v2)
 MapReduce模型的最大好處是簡便性，用戶只需要提供這兩個接口就可以處理大規模的數據，而不需要太多分布式計算的實現細節。
系統實現
 MapReduce的實時運行主要是為并行化和并發執行服務的。為了盡可能的并行化和擴展系統，MapReduce把輸入的數據分割到多個機器上。中間數據的傳輸和序列化處理等由系統來控制。分割的數據由多個Reduce來處理。這兩個步驟中Map任務和Reduce都可以同時執行，且它們都具有良好的可擴展性，也即可以方便地增加機器增加并發度。
 而在系統實現的層面上，系統需要決定底層的各個細節如數據單元的大小、中間數據的處理、內存的緩存多大、排序的方式、各個任務的調度、機器的失敗和容錯處理等等。系統自動的把這些細節都掩蓋，所以對程序員來說，他只需要知道這個編程模型并編寫MapReduce的程序即可。
 Google的論文中描述了他們在分布式機群系統上對MapReduce的實現。系統把輸入數據劃分為M份數據片，這些輸入數據片可以在不同的機器上并發的被Map函數處理。所有的中間結果對使用一個分區函數（partitioning function）分為R份。然后，對于每個分區，通過排序把具有相同鍵的所有（鍵，值）對聚合到一起，用Reduce函數處理，最后產生R個輸出文件。R的值和分區函數可以由用戶指定，系統默認的分區函數是hash(key) mod R1。
 Google的MapReduce實現是構建在GFS之上的，所有的MapReduce程序的輸入和輸出都是存儲在GFS中的文件。由于GFS中的數據都有多個副本，當執行MapReduce的機群和運行GFS的機群是同一個時，MapReduce庫的調度模塊會盡量把map任務分配到存儲數據的機器上本地運行，這樣可以避免輸入數據的網絡傳輸，極大的提高性能。此外，用戶可以指定函數用來把原始輸入數據轉換為map函數的輸入，用戶也可以指定函數用來把reduce的輸出結果序列化為輸出數據。體系結構圖1如下：

 數據的流圖[  Colby Ranger, Ramanan Raghuraman, A. P. G. B. C. K. eva luating mapreduce for multi-core and multiprocessor systems. proceedings of HPCA, 2007]如下，Worker分別執行本地的任務，可能是Map任務、Transfer任務和Reduce任務。整個過程由Master控制和協調調度。
 
 
 
 
 
 
 
 
 
 
Tplatform的實現
 我們實現的類似平臺是Tplatform。我們自己也實現了一個自制的MapReduce，建立在GFS類似的分布式文件系統TFS上[  Zhifeng Yang, Qichen Tu, K. F, L. Z, R. C, B.P. Performance gain with variable chunk size in gfs-like file systems. Journal of Computational Information System 3(2008), 1077-1084]。TFS在設計上與GFS的不同之處在于對底層Chunk大小的設置7。 與Google提供運行時庫然后通過一個二進制程序的多個副本扮演不同角色的方式不同，我們的實現提供的是一個執行MapReduce作業的服務，用戶把編寫好的實現指定接口的動態鏈接庫用系統提供的API提交上來，MapReduce系統就會自動調度和運行相關的任務。服務由一臺主控（Master）機器和若干臺工作機（Worker）組成，Master負責把用戶提交的作業（Job）切分為若干個任務，然后調度他們在各臺工作機上執行。相比提供運行時庫由用戶編譯為一個程序的方式，這樣做的好處是，系統的改進升級對用戶是不可見的。如果系統的實現改變了，只要MapReduce API不改變，用戶無需改變代碼甚至不需要重新編譯生成動態鏈接庫就可以執行MapReduce作業，這給我們未來系統的升級優化帶來了極大的便利。不僅如此，在Google的原始實現中，如果同一個機群有多個作業在同時運行，因為作業由主控程序負責調度但一個作業的主控程序是不知道另一個作業的存在的，所以多個作業之間可能產生資源的互相搶占。而在我們的系統中，一個機群只有一個主控程序，主控程序可以綜合各個作業的情況對所有任務整體進行調度。
 這里需要詳細說明我們任務設計的細節。
 我們把Worker需要做的任務分成三個類型：Map、Transfer、Reduce，我們把傳輸任務從原來的Reduce中抽離出來并作為一個可以由Worker單獨調度執行的任務。在這里我們對此設計有如下的分析。
 在原來的Map、Reduce任務的執行流程和設計下，對于Map執行完生成的中間數據，是由Reduce來到Map機器上通過遠程調用取得。這些有可能出現的場景是很多Reducer同時來一臺Map機器上進行取數據操作，造成Map機器對硬盤的隨機寫，而隨機寫對性能的影響是很大的，這樣的數據傳輸模型可以稱之為“拉”。而我們把傳輸任務獨立開來，由Master調度控制，可以控制Mapper傳輸的時間，同時Reducer在同時接到多個傳輸任務的數據時可以做緩存，避免隨機寫的出現。
 此外，我們在Worker端通過心跳線程和Master通信，在執行分配的任務時用Exec方式啟動一個新的進程來執行具體的Map和Reduce任務。而傳輸任務使用啟動線程用Socket進行傳輸。
 我們在此基礎上，實現了MapReduce的系統，我們的設計在實現上有很多和Google不同之處，也不同于開源的Hadoop2。在完成原型的開發和測試后，針對性能和系統的評估成為了我們亟待解決的關鍵問題。我們由此開始系統地對MapReduce和類似相關系統進行分析和評估，我們相信對于MapReduce和類似系統的研究工作的下一步將是對此類系統的優化。
 所以對當前系統的分析和評估成為關鍵，找到系統使用中的瓶頸所在，針對用戶需求的目標進行改進，都是實際應用中的重要問題。我們在Tplatform的實現基礎上，開發了一系列的基準程序，細致地分析了系統中可能出現的問題。
 
 系統評估
 我們總結了如下的一系列標準，用以衡量并行計算框架系統的各方面性能表現。
評估目標
單任務延遲
總機器時間
平均結束時間
加速比
公平性
故障恢復穩定性
 單任務延遲主要衡量單個任務在提交到得到響應（成功、失敗或者取消）的時間，考慮的是系統和用戶的交互能力。低延遲可以提高系統和用戶的交互能力，有利于用戶更快地知道提交任務后的結果。
 總機器時間主要衡量在系統運行過程中所有機器的用時，考慮的是整個系統的計算能力。對于同樣的任務和數據，更少的總機器時間說明了系統能有更少的計算資源（機器和時間）去完成同樣的事情。
 平均結束時間主要從多個任務的完成情況來考慮系統的性能。由于系統的目標在于統籌整個數據中心的計算資源，所以在多個任務的并行運行的情況下系統的吞吐量是值得關注的對象。
 加速比考慮系統對計算任務的加速比，主要衡量系統的擴展性能力，計算不同規模的機群節點下對任務完成情況的提高比率。
 公平性考慮在多任務同時運行的情況下，衡量系統對待各個任務的公平性。對于一些任務系統理應優先執行，而對于一些任務系統應該延后執行。由于公平性是一個比較寬泛且見仁見智的問題，我們只是針對此問題提出了一些基本的任務執行場景和評估方法。
 故障恢復穩定性衡量的是系統的故障恢復能力和穩定性。眾所周知的是，MapReduce這樣的系統通常需要運行在超大規模的集群上，故障是同種類型的系統需要處理的正常問題，所以在故障下的恢復和穩定性應該成為此類系統的評估目標。
基準程序和數據
 數據庫領域中設計了非常成功的TPC-*程序作為測試和調優的基準程序，但是在分布式計算引擎中，并沒有公認的代表程序集作為基準程序。一個優秀的基準程序集合對于系統的性能調優、不同系統的對比、調度的決策、機群的管理和資源的利用等等都應該給出統一和明確有效的衡量標準。
 基于如上考慮和系統調優的現實需求，同時考察了同類系統的測量程序集，我們初步給出如下程序集作為基準程序，并設計了一系列的指標作為衡量參數。
 基準程序集合
 基準程序集合應該是以能夠代表系統的應用程序為目標，最有代表性的是真正運行于實際系統的程序集合。但是選擇基準程序集合卻只能盡量挑選關鍵程序，使得它們在各方面的指標上都具有代表性。
 我們選取了下列的程序作為基準程序，它們主要涵蓋了應用程序的關鍵領域：搜索引擎的重要應用（PageRank、 Reverse Index）、日常分析和統計（Word Count、 String Match）、科學計算（矩陣乘法）、Web Mining（Kmeans、 Linear Regression）、衛星圖像處理（Histogram）以及典型的Map和Reduce兩階段處理的計算（PennySort）。同時我們使用不同大小的數據集來考慮系統的數據局部性和擴展性的能力，并考慮把單機的性能結果作為加速比測試的基準線。
Word Count
 Word Count也是一個MapReduce的典型處理過程，它對在一些數據中出現的詞匯進行計數。在我們的測試中，我們在CWT200G中文網頁數據集上做詞頻統計計算，此實驗用來測試系統處理大規模數據時的可靠性和穩定性。map函數分析每個HTML網頁，去除HTML標簽，進行中文分詞，以每個中文詞為鍵，值為1輸出鍵值對。reduce函數把聚合到一起的各個值加起來得到總的詞頻。程序允許是要做本地合并（local combine），這樣可以極大的減少網絡傳輸的數據量。
PennySort
 PennySort是Jim Gray提出來的一個benchmark程序[  Gray, J. Sort benchmark home page. http://research.microsoft.com/ research/barc/SortBenchmark/default.htm. ]。實驗中隨機生成長度為100字節的記錄，要求對其進行排序。PennySort使用一個MapReduce來完成，其程序的處理過程也是MapReduce的典型處理。先由Map函數提取記錄的前10個字節作為鍵，剩余記錄作為值輸出鍵值對。而Reduce函數接受分割后的數據作為輸入，進行排序。而分區函數根據鍵的范圍進行分區，可以進行區段分割，從而在各個Reduce完成后完成了整個數據的排序過程。所以這個排序程序也可以作為MapReduce系統的基準測試程序。
PageRank
 有很多鏈接分析技術用來對Web網絡結構進行分析。Pagerank是其中最重要的一個，它描述了一個網頁的重要程度[  Brin, S., and Page, L. The anatomy of a large-scale hypertextual (Web) search engine. proceedings of WWW, 1998.]，它被認為可以極大的提高搜索引擎檢索結果的精度。一個網頁的PageRank的計算公式如下：
 PR(A)=(1-d)+d(PR(T1)/C(T1)+...+PR(Tn)/C(Tn)) 
 其中，假設網頁A有n個網頁T1...Tn指向它，$C(A)$表示網頁A向外的鏈接的數量。d是一個取值在0到1之間的阻尼因子，一般取0.85。根據這個公式，網頁的PageRank可以通過若干輪迭代得到，并且可以證明迭代會收斂。
 使用MapReduce計算PageRank的過程分為兩個階段。第一階段，構建鏈接圖。map函數分析一個網頁，把網頁URL作為鍵，改網頁初始PageRank值，以及它所包含所有指出的URL作為值輸出，即URL -> (PR(URL),URL_1,URL_2,...,URL_n)。這個過程可以和前面所述的任何網頁分析的map過程結合到一起進行。reduce函數不做處理，把輸入直接作為輸出。
 第二個階段，以第一階段得到的鏈接圖作為輸入，以下MapReduce計算迭代若干次直到收斂。map函數把輸入記錄映射為(URL-> URL_1, URL_2, ..., URL_n)和 (URL_1-> PR(URL)/C(URL)), (URL_2-> PR(URL)/C(URL))等。在reduce階段，按照公式把同一個URL的各個值求和得到網頁的新的PageRank值，在按照原輸入的格式輸出。之后，用一個串行程序比較上一輪得到的PageRank值和本輪得到的值是否已經收斂，如果未收斂，則進入下一輪迭代。
 在我們的實驗中，我們使用的數據模擬Web的鏈接分布，我們生成若干個URL，鏈出URL的個數依Zip-f分布。
KMeans
 KMeans程序是數據挖掘中的基礎算法。它要求把一個點集聚合成若干個簇，簇間的點盡可能的聚合在一起。對于是用MapReduce來處理此程序，需要多次的迭代過程，所以KMeans程序也可以代表需要不斷迭代的程序集合。
 在每一次的迭代過程中，Map任務讀入點集數據中的一個子集，并把中心點向量讀入，以中心點向量來判斷每一個點是否應該輸入那個中心下的簇。它計算出每個點和中心點向量中的距離并把點分派給最近的簇。而Reduce任務把相同簇的點都聚合到一起，然后重新計算它們的中心點，然后更新中心點向量。
 一輪迭代結束后，KMeans程序會繼續迭代直到滿足收斂條件為止。收斂的條件一般為兩輪對比各個簇沒有大的變化就可以停止迭代。
Matrix Multiplication
 矩陣乘法是一個計算密集型的任務。Map任務計算輸出矩陣中的一個行，然后對矩陣中的每個元素返回位置和值。Reduce收集數據并輸出。
 評估目標
 對于上述選定的基準程序集合，我們從MapReduce的模型和系統實現出發，考慮它們的各項屬性。從而可以通過這些屬性來對基準程序進行評估，知道它們對MapReduce系統的影響是什么類型的。
 從另一方面來說，如果有新的基準程序加入，同樣考慮對這些數據屬性進行評估，就可以知道此程序的性質，如是否是數據密集型，是否導致MapReduce系統的傳輸成為瓶頸等等。
 我們考慮這些指標，如果有特殊領域的應用程序，那可以根據這些屬性值集合建立對應的基準程序，從而方便地對系統進行評估。
 基準程序的衡量指標如下[  Berkeley, Hadoop Benchmark. http://radlab.cs.berkeley.edu/wiki/Projects/Hadoop_Benchmark/Data]：
任務大小：輸入的數據字節數，Map和Reduce任務的數目。
Map任務的選擇度：在平均的Map任務中，輸出的字節數除以輸入的字節數
Reduce任務的選擇度：在平均的Reduce任務中，輸出的字節數除以輸入的字節數
Map任務的平均字節計算時間：在Map任務中計算一個字節需要的平均時間
Reduce任務的平均字節計算時間：在Reduce任務中計算一個字節需要的平均時間
數據的壓縮率：分布式文件系統中該基準程序的數據壓縮率。（注：在我們的實驗中暫不考慮此項屬性）
Map的方式：是選擇一部分數據還是對Chunk中所有數據進行順序讀入。
傳輸的方式：在進行傳輸任務時時候有偏移，對數據的分割狀況是怎么樣的，是否做到分割上的負債均衡。
中間數據大小：傳輸的時候中間數據的字節數。
Reduce的參數：是否需要數據做外排。
Map任務的復雜度：比如，為O(n)
Reduce任務的復雜度：比如，為O(n)
 我們選擇能夠代表典型MapReduce過程的PennySort，來實例說明基準程序在我們的實驗系統上各指標的值。
 實驗數據是50M條記錄的PennySort，數據量為4.8G。
 
 
 
 表格 1-50M PennySort系統評估
任務大小 輸入4.8G，Map任務75個，Reduce14個
Map任務的選擇度 1
Reduce任務的選擇度 1
Map任務的平均字節計算時間 1.72621E-06 Seconds
Reduce任務的平均字節計算時間 7.63296E-07 Seconds
數據的壓縮率 暫不考慮
Map的方式 對Chunk所有數據順序讀入
傳輸的方式 均勻分布
中間數據大小 4.23G
Reduce的參數 不需要外排
Map任務的復雜度 O(n)
Reduce任務的復雜度 nO(logn)
 
 我們說明和分析表中的數值。
 首先，Map和Reduce的任務的選擇度都是1，因為對于PennySort來說，Map做的是把數據簡單地讀入，然后進行傳輸和分割，而對Reduce來說，進行完數據的排序后也只需要把數據簡單地輸出，所以選擇度都是1.
 然后，對于傳輸的方式，按記錄的生成原則，可以均稱地進行hash分割。中間數據比初始讀入的數據反而小是因為很多數據Map任務做完后可以在本地直接進行Reduce，利用的數據的空間數據性，所以傳輸數據變小。
 最后Reduce任務需要進行排序，系統實現使用快排，復雜度為nO(logn).
 系統監控和程序概要分析
 更好地理解和監控云計算的基礎設施系統如MapReduce是一個煩人且亟待解決的問題。現有的實現都是比較簡單地記錄系統的相關性能信息，而且并沒有太多關于在此類系統中如何監控和評估的工作。但是在我們的開發和使用過程中，我們發現了系統的性能概要分析很重要，或者說通過更好地理解底層系統，能夠更好地改善和優化現有的系統。例如如下的幾個場景中，我們將說明這一點:
 數據中心中的一個程序員向系統提交了一個用高層語言如Pig Latin描述的任務后，他/她可能想知道他的任務做到什么程度。從性能概要分析的角度來考慮任務監控這個問題，任務在多個機器上的性能分布很重要。這樣可以知道任務中最耗時的函數，從來讓程序員可以針對此考慮改進自己的程序，或者在系統對任務的編譯中進行優化。
 失效在數據中心里面是正常的1。MapReduce這樣的系統對用戶掩蓋機器的失效，如果機器發生宕機，系統將處理并調度計算重執行；而對于計算任務的失效，處理方式是重新執行，如果多次失效超過一定次數，將放棄執行。這是因為在數據中心中， 很有可能是用戶提交的任務的程序中存在BUG，或者是數據有不滿足格式而導致無法讀入等等。對于需要進行長任務處理的工作來說，在現有系統的實現下，可能是一件極消耗用戶程序員精力的事情。可能的情形是，執行了很久到快結束的時候由于BUG或者存儲的問題導致失敗而最終放棄。 而實時的監控和交互可以部分地解決這個問題，讓用戶及時地知道系統里面發生的情況，對于系統無法做出判斷的事情（程序有錯），交給用戶去解決不失為一個可行的方案。
 分布式系統中的一個很重要的措施就是要保證負載均衡，這對于并行計算的框架來說，同樣意義重大。在計算的過程中記錄性能信息和進行監控，可以通知用戶或者系統。通過重新的調度或者其他手段使得負載盡可能均衡。
 總之，通過監控和程序的性能概要分析，我們可以讓系統和用戶之間有更多交互。同時給出的數據可以幫助用以評估系統，提供給不同的人如用戶或者系統開發人員分析。
實現細節
 我們需要記錄一個子任務的運行時性能概要信息，通過以下的數據結構來實現。
         struct ProfileInfo
         {
             // for map task
             int mapFanIn;
             int mapFanOut;
             int mapRecordNumber;
             int localCombineFanIn;
             int localCombineFanOut;
             int localCombineRecordNumber;
             // for reduce task
             int reduceFanIn;
             int reduceFanOut;
             int reduceRecordNumber
             // for transfer task
             int transferIO;
             int transferRecordNumber;
             // cost time, by seconds
             int taskCostTime;
         };
 對于Map階段，分別記錄扇入扇出的數據大小、map的記錄個數；以及做localcombine的扇入扇出、記錄個數；對于Reduce階段，記錄扇入扇出的數據大小、reduce的記錄個數；還有傳輸任務的傳輸數據量；最后是各個任務的花費時間。
 通過在Worker端執行任務后記錄下任務的性能概要情況，然后通過文件管道傳遞給Worker的心跳進程，然后通過心跳捎帶給Master以供分析。
 進行捎帶處理的心跳使用rpc實現，具體實現如下。
 先使用ICE的slice描述rpc的接口。
             /**
              * report to the master the task is successfully completed.
              *
              * @param taskID
              * @param profileInfo, send the profileInfo piggybackly
              */
             idempotent void completeTask(Address workerAddress, int taskID, ProfileInfo taskProfile);
 
 然后經過ICE的編譯后生成服務器端和客戶端的C++代碼，然后把任務的性能概要信息發送給Master端。
 

 評估實驗
 在這一章中，我們將對上一章中設定的系統性能指標進行評估。并闡述每一項實驗的環境、應用程序和結果分析。
機群配置
 我們的機群配置如下。
 我們在后備任務策略的評估實驗中使用了一臺Master、十四臺Worker組成的MapReduce系統集群。所有的機器都是Dell 2850服務器，每臺機器配置為2顆Intel Xeon處理器，2GB內存，6個7200 rpm SCSI硬盤組成一個RAID-0的邏輯卷。這些機器存放在兩個機架中，各有一臺Dell 2748 1Gbps交換機，機器通過一個1Gbps的全雙工以太網卡與交換機相連接，兩個機架通過一個Cisco千兆路由器鏈接。
實驗結果
單任務延遲和總機器時間
 我們使用的工作負載的數據規模如下：
WordCount，1.2G大小的數據，使用LocalCombine。
PennySort，50M條記錄共4.8G大小的數據。
PageRank，150W條URL共1.5G大小的數據。
我們實驗所得到的單任務延遲和總機器時間如下：
 表格 2-延遲和總機器時間
 Type/Time secs Latency Total Machine
WordCount 132 2013
PennySort 270 4789
PageRank 140 727
 其中，單任務延遲為用戶提交任務到任務完全結束所用時間。而總機器時間為提交任務的各個子任務（包括Map、Reduce、Transfer三種任務）的完成時間之和，度量的是對于整個機群來說的總機器時間。
平均結束時間
 我們使用上一節中的三個評估任務，同時提交給系統，并得到平均的結束時間。用以衡量在一段時間內，系統對多個任務的吞吐量。我們以平均結束時間來進行評估。
 經過實驗得到三個任務的平均結束時間為212秒，而由上一節數據可知三任務的平均延遲為180.6秒，所以我們可以通過此項評估來考慮系統是否能夠對一批任務進行優化處理。我們對我們的系統進行分析和實時監控，發現之所以慢于平均延遲，是因為對于Word Count和PageRank的一些Map被安排到比較靠后的位置執行，雖然機群中有空閑的機器，但是整個系統需要等待這些Map任務執行完后才能執行Reduce任務，從而增加了延遲。
 這也使得我們考慮后備任務的策略和更加合理的調度，使得空閑的資源能夠充分被利用，改善這些系統的評估目標。
加速比
 加速比和系統的可擴展性是MapReduce和類似系統的一個很重要的特性，正是因為非常良好的可擴展性，才使得MapReduce和其他的分布式系統區別開來，因為MapReduce系統可以很好地部署在超大規模的機群上。
 在本節的實驗里面，我們從兩方面來考察系統的可擴展性。
 第一個實驗測試在同一個規模的輸入數據和相同的配置下，Worker的增加對提交任務的延遲的影響。我們限制每臺機器可以同時運行的任務是3，傳輸任務的限制是2。我們針對5.2.1中的工作負載進行實驗，從圖中可以看到，運行的任務延遲隨Worker的增加而降低，說明此系統有良好的加速比。

 第二個實驗測試在不同的規模和相同的配置下進行，Worker的增加和數據規模成同樣的比例。從圖中可以看到，運行的任務延遲基本保持同樣的水平，表明此系統有良好的可擴展性。我們的數據規模分別為：
WordCount，1.2G、2.4G、3.6G
PennySort，50M條記錄共4.8G、100M條記錄共9.6G、150M條記錄共14.5G。
PageRank，150W條URL共1.5G、300W條URL共3.4G、450W條URL共6G。
 注意PageRank由于相互間鏈接增加的原因數據規模增加斜率大于線性增加。這三個不同大小的數據集合分別在5、10、14臺機器上運行。
 實驗結果如下：
公平性
 對于公平的定義，在不同的應用場合有不同的評估方法，我們在這一節的評估中，簡單地先考慮一種場景，并評估我們系統的公平性。
 我們進行如下的實驗：先提交一個長任務，然后過一段時間提交一個短任務。評估系統的調度對此短任務來說是否公平。
 我們準備的長任務是500M條記錄的PennySort，數據規模為50G，在我們以前的實驗中，我們的系統大約需要2900秒才能完成此任務，它屬于長任務。同時我們準備了一個短任務，是10M條記錄的PennySort，數據規模是960M，在我們以前的實驗中大約只需要50秒就能完成。這里我們使用的任務類型是一樣的，都是做排序，我們僅僅考慮任務的完成時間對公平性的影響，在實際應用中可能還會考慮提交任務的權重等等，這些具體的應用不是我們考慮的范圍。這里長任務我們記為L（long）任務，短任務我們記為S（short）任務。
 通過實驗我們發現，由于S任務的很多子任務沒有得到及時調度，在S任務提交后，經過356秒才完成了S任務，而最后L任務的延遲為2791秒，基本沒有受到短任務的影響。但是由于調度的不合理，對于S任務來說調度是不公平的，它提交了很長一段時間后部分子任務才得到處理。
故障恢復穩定性
 在分析和測試評估MapReduce和類似系統時，一個重要的方面就是這些系統的容錯性，因為各種故障在這樣的系統中是屬于正常情況的。
 我們通過實驗模擬各種故障的發生：如殺死Worker進程模擬宕機、硬盤寫滿或其他模擬硬盤出錯、突然中斷一些Worker之間的網絡通信等等。利用這些模擬的實驗來評估系統的穩定性。
 實驗表明，一個高穩定性的系統才能在這樣的環境中良好地工作。我們系統的穩定性也是未來工作的一個方向。
實驗結果和性能問題分析
 我們通過實驗和分析系統，發現了一些系統的性能問題。列舉一些我們覺得目前可能成為瓶頸的如下：
一些機器相較別的機器的慢速，成為落后者，會極大地增加任務完成的延遲。
 由下圖可以看到，大部分的任務完成時間趨同，而個別任務顯著地比其他任務慢，導致最終的延遲降低，成為落后者。這就是落后者的問題表現。

首頁 上一頁 1 2 下一頁 尾頁 1/2/2