Hadoop框架的理解

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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