1、長沙引擎信息技術(shù)有限公司Hadoop框架理解劉勇2014-02-16本文從基本架構(gòu)、HDFS及MapReduce基本工作原理上對hadoop框架進(jìn)行了粗略的分析，可幫助初學(xué)者初步理解hadoop架構(gòu)。目 錄Hadoop框架的理解11、Hadoop架構(gòu)11.1、簡介：11.2、優(yōu)點(diǎn)：21.3、架構(gòu)：22、HDFS（Hadoop Distributed FileSystem）32.1、簡介：32.2、文件分塊：42.3、NameNode和DataNode：42.4、數(shù)據(jù)流53、MapReduce73.1簡介：73.2工作原理73.3實例：93.4 Shuffle（洗牌）103.5 MapReduc

2、e執(zhí)行過程143.6 MapReduce 適合處理的任務(wù)16Hadoop框架的理解1、 Hadoop架構(gòu)1.1、簡介：Hadoop是一個分布式系統(tǒng)基礎(chǔ)架構(gòu)，由Apache基金會所開發(fā)。用戶可以在不了解分布式底層細(xì)節(jié)的情況下，開發(fā)分布式程序。充分利用集群的威力高速運(yùn)算和存儲。Hadoop包括兩個核心部分：Hadoop分布式文件系統(tǒng)（Hadoop Distributed File System，HDFS）和MapReduce編程模型。其中HDFS運(yùn)行在商用硬件上，它和現(xiàn)有分布式文件系統(tǒng)很相似，但也具備了明顯的差異性，比如HDFS是高度容錯的，可運(yùn)行在廉價硬件上；HDFS能為應(yīng)用程序提供高吞吐率的數(shù)

3、據(jù)訪問，適用于大數(shù)據(jù)集的應(yīng)用中；HDFS在POSIX規(guī)范進(jìn)行了修改，使之能對文件系統(tǒng)數(shù)據(jù)進(jìn)行流式訪問，從而適用于批量數(shù)據(jù)的處理。HDFS為文件采用一種一次寫多次讀的訪問模型，從而簡化了數(shù)據(jù)一致性問題，使高吞吐率數(shù)據(jù)訪問成為可能，一些Map/Reduce應(yīng)用和網(wǎng)頁抓取程序在這種訪問模型下表現(xiàn)完美。MapReduce 本身源自于函數(shù)式語言，主要通過Map（映射）和Reduce（化簡）這兩個步驟來并行處理大規(guī)模的數(shù)據(jù)集。首先，Map會先對由很多獨(dú)立元素組 成的邏輯列表中的每一個元素進(jìn)行指定的操作，且原始列表不會被更改，會創(chuàng)建多個新的列表來保存Map的處理結(jié)果。也就意味著，Map操作是高度并行的。當(dāng)M

4、ap工作完成之后，系統(tǒng)會接著對新生成的多個列表進(jìn)行清理（Shuffle）和排序，之后，會這些新創(chuàng)建的列表進(jìn)行Reduce操作，也就是對一個列表中的元素根據(jù)Key值進(jìn)行適當(dāng)?shù)暮喜ⅰ?.2、優(yōu)點(diǎn)：Hadoop是一個能夠?qū)Υ罅繑?shù)據(jù)進(jìn)行分布式處理的軟件框架。但是 Hadoop 是以一種可靠、高效、可伸縮的方式進(jìn)行處理的。Hadoop 是可靠的，因為它假設(shè)計算元素和存儲會失敗，因此它維護(hù)多個工作數(shù)據(jù)副本，確保能夠針對失敗的節(jié)點(diǎn)重新分布處理。Hadoop 是高效的，因為它以并行的方式工作，通過并行處理加快處理速度。Hadoop 還是可伸縮的，能夠處理 PB 級數(shù)據(jù)。此外，Hadoop 依賴于社區(qū)服務(wù)器，因

5、此它的成本比較低，任何人都可以使用。Hadoop是一個能夠讓用戶輕松架構(gòu)和使用的分布式計算平臺。用戶可以輕松地在Hadoop上開發(fā)和運(yùn)行處理海量數(shù)據(jù)的應(yīng)用程序。它主要有以下幾個優(yōu)點(diǎn)：A、 高可靠性。Hadoop按位存儲和處理數(shù)據(jù)的能力值得人們信賴。B、 高擴(kuò)展性。Hadoop是在可用的計算機(jī)集簇間分配數(shù)據(jù)并完成計算任務(wù)的，這些集簇可以方便地擴(kuò)展到數(shù)以千計的節(jié)點(diǎn)中。C、 高效性。Hadoop能夠在節(jié)點(diǎn)之間動態(tài)地移動數(shù)據(jù)，并保證各個節(jié)點(diǎn)的動態(tài)平衡，因此處理速度非?？臁、 高容錯性。Hadoop能夠自動保存數(shù)據(jù)的多個副本，并且能夠自動將失敗的任務(wù)重新分配。E、 低成本。與一體機(jī)、商用數(shù)據(jù)倉庫以及Q

6、likView、Yonghong Z-Suite等數(shù)據(jù)集市相比，hadoop是開源的，項目的軟件成本因此會大大降低。Hadoop帶有用Java語言編寫的框架，因此運(yùn)行在 Linux 生產(chǎn)平臺上是非常理想的。Hadoop 上的應(yīng)用程序也可以使用其他語言編寫，比如C+。1.3、架構(gòu)：Hadoop 由許多元素構(gòu)成。其最底部是 Hadoop Distributed FileSystem（HDFS），它存儲 Hadoop 集群中所有存儲節(jié)點(diǎn)上的文件。HDFS（對于本文）的上一層是MapReduce引擎，該引擎由 JobTrackers 和 TaskTrackers 組成。2、HDFS（Hadoop Di

7、stributed FileSystem）2.1、簡介：對外部客戶機(jī)而言，HDFS就像一個傳統(tǒng)的分級文件系統(tǒng)?？梢詣?chuàng)建、刪除、移動或重命名文件，等等。但是 HDFS 的架構(gòu)是基于一組特定的節(jié)點(diǎn)構(gòu)建的，這是由它自身的特點(diǎn)決定的。這些節(jié)點(diǎn)包括 NameNode（僅一個），它在 HDFS 內(nèi)部提供元數(shù)據(jù)服務(wù)；DataNode，它為 HDFS 提供存儲塊。由于僅存在一個 NameNode，因此這是 HDFS 的一個缺點(diǎn)（單點(diǎn)失?。４鎯υ?HDFS 中的文件被分成塊，然后將這些塊復(fù)制到多個計算機(jī)中（DataNode）。這與傳統(tǒng)的 RAID 架構(gòu)大不相同。塊的大小（通常為 64MB）和復(fù)制的塊數(shù)量在創(chuàng)建

8、文件時由客戶機(jī)決定。NameNode 可以控制所有文件操作。HDFS 內(nèi)部的所有通信都基于標(biāo)準(zhǔn)的TCP/IP協(xié)議。2.2、文件分塊：HDFS同樣也有塊(block)的概念，但是大得多，默認(rèn)為64 MB。與單一磁盤上的文件系統(tǒng)相似，HDFS上的文件也被劃分為塊大小的多個分塊,作為獨(dú)立的存儲單元。但與其他文件系統(tǒng)不同的是，HDFS中小于一個塊大小的文件不會占據(jù)整個塊的空間。對分布式文件系統(tǒng)中的塊進(jìn)行抽象會帶來很多好處。第一個最明顯的好處是，一個文件的大小可以大于網(wǎng)絡(luò)中任意一個磁盤的容量。文件的所有塊并不需要存儲在同一個磁盤上，因此它們可以利用集群上的任意一個磁盤進(jìn)行存儲。事實上，盡管不常見，但對于

9、整個HDFS集群而言，也可以僅存儲一個文件，該文件的塊占滿集群中所有的磁盤。第二個好處是，使用塊抽象而非整個文件作為存儲單元，大大簡化了存儲子系統(tǒng)的設(shè)計。簡化是所有系統(tǒng)的目標(biāo)，伹是這對于故障種類繁多的分布式系統(tǒng)來說尤為重要。將存儲子系統(tǒng)控制單元設(shè)置為塊，可簡化存儲管理（由于塊的大小是固定的， 因此計算單個磁盤能存儲多少個塊就相對容易）。同時也消除了對元數(shù)據(jù)的顧慮(塊只是存儲數(shù)據(jù)的一部分，而文件的元數(shù)據(jù)，如權(quán)限信息，并不需要與塊一同存儲，這樣一來，其他系統(tǒng)就可以單獨(dú)地管理這些元數(shù)據(jù))。不僅如此，塊非常適合用于數(shù)據(jù)備份進(jìn)而提供數(shù)據(jù)容錯能力和可用性。將每個塊復(fù)制到少數(shù)幾個獨(dú)立的機(jī)器上(默認(rèn)為3個)，

10、可以確保在發(fā)生塊、磁盤或機(jī)器故障后數(shù)據(jù)不丟失。如果發(fā)現(xiàn)一個塊不可用，系統(tǒng)會從其他地方讀取另一個復(fù)本，而這個過程對用戶是透明的。一個因損壞或機(jī)器故障而丟失的塊可以從其他候選地點(diǎn)復(fù)制到另一臺可以正常運(yùn)行的機(jī)器上，以保證復(fù)本的數(shù)量回到正常水平。同樣，有些應(yīng)用程序可能選擇為一些常用的文件塊設(shè)置更高的復(fù)本數(shù)量進(jìn)而分散集群中的讀取負(fù)載。2.3、NameNode和DataNode：HDFS集群有兩類節(jié)點(diǎn)，并以管理者-工作者模式運(yùn)行，即一個NameNode(管理者)和多個DataNode(工作者)。NameNode管理文件系統(tǒng)的命名空間，它維護(hù)著文件系統(tǒng)樹及整棵樹內(nèi)所有的文件和目錄。這些信息以兩個文件形式永久

11、保存在本地磁盤上：命名空間鏡像文件和編輯日志文件。NameNode也記錄著每個文件中各個塊所在的數(shù)據(jù)節(jié)點(diǎn)信息，但它并不永久保存塊的位置信息，因為這些信息會在系統(tǒng)啟動時由數(shù)據(jù)節(jié)點(diǎn)重建。DataNode是文件系統(tǒng)的工作節(jié)點(diǎn)。它們根據(jù)需要存儲并檢索數(shù)據(jù)塊(受客戶端或 NameNode調(diào)度），并且定期向NameNode發(fā)送它們所存儲的塊的列表。沒有NameNode,文件系統(tǒng)將無法使用。事實上，如果運(yùn)行NameNode服務(wù)的機(jī)器毀壞，文件系統(tǒng)上所有的文件將會丟失，因為我們不知道如何根據(jù)DataNode的塊來重建文件。因此，對NameNode實現(xiàn)容錯非常重要，Hadoop為此提供了兩種機(jī)制。第一種機(jī)制是備

12、份那些組成文件系統(tǒng)元數(shù)據(jù)持久狀態(tài)的文件。Hadoop可以通過配置使NameNode在多個文件系統(tǒng)上保存元數(shù)據(jù)的持久狀態(tài)。這些寫操作是實時同步的，是原子操作。一般的配置是將持久狀態(tài)寫人本地磁盤的同時，寫人一個遠(yuǎn)程掛載的網(wǎng)絡(luò)文件系統(tǒng)(NFS)。另一種可行的方法是運(yùn)行一個輔助NameNode,但它不能被用作NameNode。這個輔助NameNode的重要作用是定期通過編輯日志合并命名空間鏡像，以防止編輯日志過大。這個輔助NameNode 般在另一臺單獨(dú)的物理計算機(jī)上運(yùn)行，因為它需要占用大量CPU時間與NameNode相同容量的內(nèi)存來執(zhí)行合并操作。它會保存合并后的命名空間鏡像的副本，并在NameNod

13、e發(fā)生故障時啟用。但是，輔助NameNode保存的狀態(tài)總是滯后于主節(jié)點(diǎn)，所以在主節(jié)點(diǎn)全部失效時，難免會丟失部分?jǐn)?shù)據(jù)。在這種情況下，一般把存儲在NFS上的NameNode元數(shù)據(jù)復(fù)制到輔助NameNode并作為新的主NameNode運(yùn)行。2.4、數(shù)據(jù)流2.4.1、文件讀取剖析為了了解客戶端及與之交互的HDFS、NameNode和DataNode之間的數(shù)據(jù)流是什么樣的，我們可參考下圖,該圖顯示了在讀取文件時一些事件的主要順序。大致過程如下：1、使用HDFS提供的客戶端開發(fā)庫Client，向遠(yuǎn)程的NameNode發(fā)起RPC請求；2、NameNode會視情況返回文件的部分或者全部block列表，對于每個

14、block，NameNode都會返回有該block拷貝的DataNode地址；3、客戶端開發(fā)庫Client會選取離客戶端最接近的DataNode來讀取block；如果客戶端本身就是DataNode,那么將從本地直接獲取數(shù)據(jù)；4、讀取完當(dāng)前block的數(shù)據(jù)后，關(guān)閉與當(dāng)前的DataNode連接，并為讀取下一個block尋找最佳的DataNode；5、當(dāng)讀完列表的block后，且文件讀取還沒有結(jié)束，客戶端開發(fā)庫會繼續(xù)向NameNode獲取下一批的block列表。6、讀取完一個block都會進(jìn)行checksum驗證，如果讀取DataNode時出現(xiàn)錯誤，客戶端會通知NameNode，然后再從下一個擁有該

15、block拷貝的DataNode繼續(xù)讀。2.4.2、文件寫入剖析文件是如何寫入HDFS的。盡管比較詳細(xì)，但對于理解數(shù)據(jù)流還是很有用的，因為它清楚地說明了HDFS的一致模型。要考慮的情況是如何創(chuàng)建一個新文件，并把數(shù)據(jù)寫入該文件，最后關(guān)閉該文件。寫入文件的過程比讀取較為復(fù)雜：1、使用HDFS提供的客戶端開發(fā)庫Client，向遠(yuǎn)程的Namenode發(fā)起RPC請求；2、Namenode會檢查要創(chuàng)建的文件是否已經(jīng)存在，創(chuàng)建者是否有權(quán)限進(jìn)行操作，成功則會為文件創(chuàng)建一個記錄，否則會讓客戶端拋出異常；3、當(dāng)客戶端開始寫入文件的時候，開發(fā)庫會將文件切分成多個packets，并在內(nèi)部以數(shù)據(jù)隊列data queue

16、的形式管理這些packets，并向Namenode申請新的blocks，獲取用來存儲備份的合適的datanodes列表，列表的大小根據(jù)在Namenode中對replication的設(shè)置而定。4、開始以pipeline（管道）的形式將packet寫入所有的備份中。開發(fā)庫把packet以流的方式寫入第一個datanode，該datanode把該packet存儲之后，再將其傳遞給在此pipeline中的下一個datanode，直到最后一個datanode，這種寫數(shù)據(jù)的方式呈流水線的形式。5、最后一個datanode成功存儲之后會返回一個ack packet，在pipeline里傳遞至客戶端，在客戶端

17、的開發(fā)庫內(nèi)部維護(hù)著ack queue，成功收到datanode返回的ack packet后會從ack queue移除相應(yīng)的packet。6、如果傳輸過程中，有某個datanode出現(xiàn)了故障，那么當(dāng)前的pipeline會被關(guān)閉，出現(xiàn)故障的datanode會從當(dāng)前的pipeline中移除，剩余的block會繼續(xù)剩下的datanode中繼續(xù)以pipeline的形式傳輸，同時Namenode會分配一個新的datanode，保持replicas設(shè)定的數(shù)量。3、MapReduce3.1簡介：MapReduce是一種編程模型，用于大規(guī)模數(shù)據(jù)集（大于1TB）的并行運(yùn)算。概念Map（映射）和Reduce（規(guī)約）

18、，和他們的主要思想，都是從函數(shù)式編程語言里借來的，還有從矢量編程語言里借來的特性。他極大地方便了編程人員在不會分布式并行編程的情況下，將自己的程序運(yùn)行在分布式系統(tǒng)上。 當(dāng)前的軟件實現(xiàn)是指定一個Map（映射）函數(shù)，用來把一組鍵值對映射成一組新的鍵值對，指定并發(fā)的Reduce（規(guī)約）函數(shù)，用來保證所有映射的鍵值對中的每一個共享相同的鍵組。3.2工作原理在Hadoop官方文檔介紹了Hadoop中MapReduce的三個步驟：map(主要是分解并行的任務(wù))、combine(主要是為了提高reduce的效率)和reduce(把處理后的結(jié)果再匯總起來)。3.2.1、map由于map是并行地對輸入的文件集進(jìn)

19、行操作，所以它的第一步(FileSplit)就是把文件集分割成一些子集。如果單個的文件大到影響查找效率時，它會被分割成一些小的文件。要指出的是，分割這一步是不知道輸入文件的內(nèi)部邏輯結(jié)構(gòu)的。比如，以行為邏輯分割的文本文件會被以任意的字節(jié)界限分割，所以這個具體分割要由用戶自己指定。然后每個文件分割體都會對應(yīng)地有一個新的map任務(wù)。當(dāng)單個map任務(wù)開始時，它會對每個配置過的reduce任務(wù)開啟一個新的輸出流(writer)，這個輸出流會讀取文件分割體。Hadoop中的類InputFormat用于分析輸入文件并產(chǎn)生鍵值(key/value)對。Hadoop中的Mapper類是一個可以由用戶實現(xiàn)的類，經(jīng)

20、過InputFormat類分析的鍵值(key/value)對都傳給Mapper類，這樣，用戶提供的Mapper類就可以進(jìn)行真正的map操作。當(dāng)map操作的輸出被收集后，它們會被Hadoop中的Partitioner類以指定的方式區(qū)分地寫入輸出文件里。3.2.2、combine當(dāng)map操作輸出它的鍵值(key/value)對時，出于性能和效率的考慮，Hadhoop框架提供了一個合成器(combine)。有了這個合成器，map操作所產(chǎn)生的鍵值(key/value)對就不會馬上寫入輸出文件，它們會被收集在一些list中，一個key值對應(yīng)一個list，當(dāng)寫入一定數(shù)量的鍵值(key/value)對時，這

21、部分list會被合成器處理。比如，hadoop案例中的word count程序，它的map操作輸出是(word,1)鍵值對，在map操作的輸入中，詞的計數(shù)可以使用合成器來加速。合成操作會在內(nèi)存中收集處理list，一個詞一個list。當(dāng)一定數(shù)量的鍵值對輸出到內(nèi)存中時，就調(diào)用合成操作的reduce方法，每次都以一個唯一的詞為key，values是list的迭代器，然后合成器輸出(word, count in this part of the input)鍵值對。3.2.3、reduce當(dāng)一個reduce任務(wù)開始時，它的輸入分散在各個節(jié)點(diǎn)上的map的輸出文件里。如果在分布式的模式下，需要先把這些文件

22、拷貝到本地文件系統(tǒng)上。一旦所有的數(shù)據(jù)都被拷貝到reduce任務(wù)所在的機(jī)器上時，reduce任務(wù)會把這些文件合并到一個文件中。然后這個文件會被合并分類，使得相同的key的鍵值對可以排在一起。接下來的reduce操作就很簡單了，順序地讀入這個文件，將鍵(key)所對應(yīng)的值(values)傳給reduce方法完成之后再讀取一個鍵(key)。最后，輸出由每個reduce任務(wù)的輸出文件組成。而它們的格式可以由JobConf.setOutputFormat類指定。3.3實例：在map階段輸入的是原始的NCDC數(shù)據(jù)。我們選擇的是一種文本輸入格式，以便數(shù)據(jù)集的每一行都會是一個文本值。鍵是在文件開頭部分文本行起

23、始處的偏移量，但我們沒有這方面的需要，所以將其忽略。map函數(shù)很簡單。我們使用map函數(shù)來找出年份和氣溫，因為我們只對它們有興趣。在本例中，map函數(shù)只是一個數(shù)據(jù)準(zhǔn)備階段，通過這種方式來建立數(shù)據(jù)，使得reducer函數(shù)能在此基礎(chǔ)上進(jìn)行工作：找出每年的最高氣溫。選用其中的幾行數(shù)據(jù)進(jìn)行說明： .N9+00001+. .N9+00221+. .N9-00111+. .N9+01111+. .N9+00781+.這些行以鍵/值對的方式來表示map函數(shù)：(0,.N9+00001+.) (106,.N9+00221+.) (212,.N9-00111+.) (318,.N9+01111+.) (424,.

24、N9+00781+.)鍵是文件中的行偏移量，而這往往是我們在map函數(shù)中所忽視的。map函數(shù)的功能僅僅提取年份和氣溫(以粗體顯示)，并將其作為輸出被發(fā)送。(氣溫值已被解釋為整數(shù))(1950,0)(1950,22)(1950, 11)(1949,111)(1949,78)map函數(shù)的輸出先由MapReduce框架處理，然后再被發(fā)送到reduce函數(shù)。這一處理過程根據(jù)鍵來對鍵/值對進(jìn)行排序和分組。因此，繼續(xù)我們的示例，reduce 函數(shù)會看到如下輸入：(1949,111,78)(1950,0,22,11)每年的年份后都有一系列氣溫讀數(shù)。所有reduce函數(shù)現(xiàn)在必須重復(fù)這個列表并從中找出最大的讀數(shù)：

25、(1949,111)(1950,22)這是最后的輸出：全球氣溫記錄中每年的最高氣溫。整個數(shù)據(jù)流如圖所示：3.4 Shuffle（洗牌）Shuffle在MapReduce中是一個核心過程，它在接點(diǎn)中的數(shù)據(jù)交換起著關(guān)鍵的作用，此過程橫跨map與reduce兩端。3.4.1 Map端整個流程大致分四步。簡單些可以這樣說，每個map task都有一個內(nèi)存緩沖區(qū)，存儲著map的輸出結(jié)果，當(dāng)緩沖區(qū)快滿的時候需要將緩沖區(qū)的數(shù)據(jù)以一個臨時文件的方式存放到磁盤，當(dāng)整個map task結(jié)束后再對磁盤中這個map task產(chǎn)生的所有臨時文件做合并，生成最終的正式輸出文件，然后等待reduce task來拉數(shù)據(jù)。當(dāng)然

26、這里的每一步都可能包含著多個步驟與細(xì)節(jié)，下面對細(xì)節(jié)來一一說明：1. 在map task執(zhí)行時，它的輸入數(shù)據(jù)來源于HDFS的block，當(dāng)然在MapReduce概念中，map task只讀取split。Split與block的對應(yīng)關(guān)系可能是多對一，默認(rèn)是一對一。2. 在經(jīng)過mapper的運(yùn)行后，我們得知mapper的輸出是這樣一個key/value對，到底當(dāng)前結(jié)果應(yīng)該交由哪個reduce去做呢，是需要現(xiàn)在決定的。MapReduce提供Partitioner接口，它的作用就是根據(jù)key或value及reduce的數(shù)量來決定當(dāng)前的這對輸出數(shù)據(jù)最終應(yīng)該交由哪個reduce task處理。默認(rèn)對key

27、hash后再以reduce task數(shù)量取模。默認(rèn)的取模方式只是為了平均reduce的處理能力，如果用戶自己對Partitioner有需求，可以訂制并設(shè)置到j(luò)ob上。3. 環(huán)形內(nèi)存緩沖區(qū)是有大小限制的，默認(rèn)是100MB。當(dāng)map task的輸出結(jié)果很多時，就可能會撐爆內(nèi)存，所以需要在一定條件下將緩沖區(qū)中的數(shù)據(jù)臨時寫入磁盤，然后重新利用這塊緩沖區(qū)。這個從內(nèi)存往磁盤寫數(shù)據(jù)的過程被稱為Spill，中文可譯為溢寫，字面意思很直觀。這個溢寫是由單獨(dú)線程來完成，不影響往緩沖區(qū)寫map結(jié)果的線程。溢寫線程啟動時不應(yīng)該阻止map的結(jié)果輸出，所以整個緩沖區(qū)有個溢寫的比例spill.percent。這個比例默認(rèn)是

28、0.8，也就是當(dāng)緩沖區(qū)的數(shù)據(jù)已經(jīng)達(dá)到閾值（buffer size * spill percent = 100MB * 0.8 = 80MB），溢寫線程啟動，鎖定這80MB的內(nèi)存，執(zhí)行溢寫過程。Map task的輸出結(jié)果還可以往剩下的20MB內(nèi)存中寫，互不影響。當(dāng)溢寫線程啟動后，需要對這80MB空間內(nèi)的key做排序(Sort)。排序是MapReduce模型默認(rèn)的行為，這里的排序也是對序列化的字節(jié)做的排序。在這里我們可以想想，因為map task的輸出是需要發(fā)送到不同的reduce端去，而內(nèi)存緩沖區(qū)沒有對將發(fā)送到相同reduce端的數(shù)據(jù)做合并，那么這種合并應(yīng)該是體現(xiàn)是磁盤文件中的。從官方圖上也可以

29、看到寫到磁盤中的溢寫文件是對不同的reduce端的數(shù)值做過合并。所以溢寫過程一個很重要的細(xì)節(jié)在于，如果有很多個key/value對需要發(fā)送到某個reduce端去，那么需要將這些key/value值拼接到一塊，減少與partition相關(guān)的索引記錄。在針對每個reduce端而合并數(shù)據(jù)時，有些數(shù)據(jù)應(yīng)該把它們的值合并到一塊，這個過程叫reduce也叫combine。但MapReduce的術(shù)語中，reduce只指reduce端執(zhí)行從多個map task取數(shù)據(jù)做計算的過程。除reduce外，非正式地合并數(shù)據(jù)只能算做combine了。其實大家知道的，MapReduce中將Combiner等同于Reduce

30、r。如果client設(shè)置過Combiner，那么現(xiàn)在就是使用Combiner的時候了。將有相同key的key/value對的value加起來，減少溢寫到磁盤的數(shù)據(jù)量。Combiner會優(yōu)化MapReduce的中間結(jié)果，所以它在整個模型中會多次使用。那哪些場景才能使用Combiner呢？從這里分析，Combiner的輸出是Reducer的輸入，Combiner絕不能改變最終的計算結(jié)果。所以從我的想法來看，Combiner只應(yīng)該用于那種Reduce的輸入key/value與輸出key/value類型完全一致，且不影響最終結(jié)果的場景。比如累加，最大值等。Combiner的使用一定得慎重，如果用好，它

31、對job執(zhí)行效率有幫助，反之會影響reduce的最終結(jié)果。4. 每次溢寫會在磁盤上生成一個溢寫文件，如果map的輸出結(jié)果真的很大，有多次這樣的溢寫發(fā)生，磁盤上相應(yīng)的就會有多個溢寫文件存在。當(dāng)map task真正完成時，內(nèi)存緩沖區(qū)中的數(shù)據(jù)也全部溢寫到磁盤中形成一個溢寫文件。最終磁盤中會至少有一個這樣的溢寫文件存在(如果map的輸出結(jié)果很少，當(dāng)map執(zhí)行完成時，只會產(chǎn)生一個溢寫文件)，因為最終的文件只有一個，所以需要將這些溢寫文件歸并到一起，這個過程就叫做Merge。因為merge是將多個溢寫文件合并到一個文件，所以可能也有相同的key存在，在這個過程中如果client設(shè)置過Combiner，也會

32、使用Combiner來合并相同的key。至此，map端的所有工作都已結(jié)束，最終生成的這個文件也存放在TaskTracker夠得著的某個本地目錄內(nèi)。每個reduce task不斷地通過RPC從JobTracker那里獲取map task是否完成的信息，如果reduce task得到通知，獲知某臺TaskTracker上的map task執(zhí)行完成，Shuffle的后半段過程開始啟動。3.4.2 Reduce端簡單地說，reduce task在執(zhí)行之前的工作就是不斷地拉取當(dāng)前job里每個map task的最終結(jié)果，然后對從不同地方拉取過來的數(shù)據(jù)不斷地做merge，也最終形成一個文件作為reduce

33、task的輸入文件。Shuffle在reduce端的過程也能用圖上標(biāo)明的三點(diǎn)來概括。當(dāng)前reduce copy數(shù)據(jù)的前提是它要從JobTracker獲得有哪些map task已執(zhí)行結(jié)束，這段過程不表，有興趣的朋友可以關(guān)注下。Reducer真正運(yùn)行之前，所有的時間都是在拉取數(shù)據(jù)，做merge，且不斷重復(fù)地在做。如前面的方式一樣，下面我也分段地描述reduce 端的Shuffle細(xì)節(jié)： 1. Copy過程，簡單地拉取數(shù)據(jù)。Reduce進(jìn)程啟動一些數(shù)據(jù)copy線程(Fetcher)，通過HTTP方式請求map task所在的TaskTracker獲取map task的輸出文件。因為map task早

34、已結(jié)束，這些文件就歸TaskTracker管理在本地磁盤中。2. Merge階段。這里的merge如map端的merge動作，只是數(shù)組中存放的是不同map端copy來的數(shù)值。Copy過來的數(shù)據(jù)會先放入內(nèi)存緩沖區(qū)中，這里的緩沖區(qū)大小要比map端的更為靈活，它基于JVM的heap size設(shè)置，因為Shuffle階段Reducer不運(yùn)行，所以應(yīng)該把絕大部分的內(nèi)存都給Shuffle用。這里需要強(qiáng)調(diào)的是，merge有三種形式：1)內(nèi)存到內(nèi)存 2)內(nèi)存到磁盤 3)磁盤到磁盤。默認(rèn)情況下第一種形式不啟用，讓人比較困惑，是吧。當(dāng)內(nèi)存中的數(shù)據(jù)量到達(dá)一定閾值，就啟動內(nèi)存到磁盤的merge。與map 端類似，這也

35、是溢寫的過程，這個過程中如果你設(shè)置有Combiner，也是會啟用的，然后在磁盤中生成了眾多的溢寫文件。第二種merge方式一直在運(yùn)行，直到?jīng)]有map端的數(shù)據(jù)時才結(jié)束，然后啟動第三種磁盤到磁盤的merge方式生成最終的那個文件。3. Reducer的輸入文件。不斷地merge后，最后會生成一個“最終文件”。為什么加引號？因為這個文件可能存在于磁盤上，也可能存在于內(nèi)存中。對我們來說，當(dāng)然希望它存放于內(nèi)存中，直接作為Reducer的輸入，但默認(rèn)情況下，這個文件是存放于磁盤中的。當(dāng)Reducer的輸入文件已定，整個Shuffle才最終結(jié)束。然后就是Reducer執(zhí)行，把結(jié)果放到HDFS上。3.5 Ma

36、pReduce執(zhí)行過程整個過程有4個獨(dú)立的實體 客戶端：提交MapReduce JobTracker：協(xié)調(diào)作業(yè)的運(yùn)行 TaskTracker：運(yùn)行作業(yè)劃分后的任務(wù) HDFS：用來在其他實體之間共享作業(yè)文件下圖為整體運(yùn)行圖：3.5.1 正常情況A.作業(yè)的提交JobClient的runJob是用于新建JobClient實例并調(diào)用其submitJob()方法的便捷方式，提交Job后，runJob()每秒輪詢檢測作業(yè)的進(jìn)度，隨時監(jiān)控Job的運(yùn)行狀態(tài)。其中JobClient的submitJob()方法所實現(xiàn)的作業(yè)提交過程： 向JobTracker請求一個新的作業(yè)ID 檢查作業(yè)的輸出說明 計算作業(yè)的輸入分

37、片 將運(yùn)行作業(yè)所需要的資源(Jar文件，配置文件和計算所得輸入分片)復(fù)制到一個作業(yè)ID命名的目錄下JobTracker的文件系統(tǒng)中。B.作業(yè)的初始化JobTracker接收對其提交的作業(yè)后，會把這個調(diào)用放入一個隊列，交由作業(yè)調(diào)度器調(diào)度，初始化。初始化包括創(chuàng)建一個表示正在運(yùn)行作業(yè)的對象-封裝任務(wù)和記錄信息，以便跟蹤任務(wù)的狀態(tài)和進(jìn)程C.任務(wù)的分配TaskTracker運(yùn)行簡單的循環(huán)來對JobTracker發(fā)送心跳，告知自己的是否存活，同時交互信息，對于map任務(wù)和reduce任務(wù)，TaskTracker會分配適當(dāng)?shù)墓潭〝?shù)量的任務(wù)槽，理想狀態(tài)一般遵循數(shù)據(jù)本地化，和機(jī)架本地化D.任務(wù)的執(zhí)行第一步：Ta

38、skTracker拷貝JAR文件到本地，第二部：TaskTracker新建本地目錄，將JAR文件加壓到其下面；第三步：TaskTracker新建一個TaskRunner實例運(yùn)行該任務(wù)。E.進(jìn)程和狀態(tài)的更新通過Job的Status屬性對Job進(jìn)行檢測，例如作業(yè)云習(xí)慣狀態(tài)，map和reduce運(yùn)行的進(jìn)度、Job計數(shù)器的值、狀態(tài)消息描述等等，尤其對計數(shù)器Counter(計數(shù)器)屬性的檢查。F.作業(yè)的完成當(dāng)JobTracker收到Job最后一個Task完成的消息時候便把Job的狀態(tài)設(shè)置為”完成“，JobClient得知后，從runJob()方法返回。3.5.2 失敗情況A.TasK失敗第一種情況:map或reduce任務(wù)中的用戶代碼拋出運(yùn)行異常，此時子進(jìn)程JVM進(jìn)程會在退出之前想TaskTracker發(fā)送錯誤報告，錯誤報告被記錄錯誤日志，TaskTracker會將這個任務(wù)(Task)正在運(yùn)行的Task Attempt標(biāo)記為失敗，釋放一個任務(wù)槽去運(yùn)行另外一個Task Attempt。第二種情況：子進(jìn)程JVM突然退出Task Tracker會注意到JVM退出，并將此Task Attempt標(biāo)記為失敗。JobTracker通過心跳得知一個Task Attempt失敗后，會重啟調(diào)度該Task的執(zhí)行，默認(rèn)情況下如果失敗4次不會重試(通過mapre