第五章MapReduce5.1MapReduce簡(jiǎn)介5.1MapReduce簡(jiǎn)介簡(jiǎn)單來(lái)說(shuō)，MapReduce是一種思想，或是一種編程模型。對(duì)Hadoop來(lái)說(shuō)，MapReduce是一個(gè)分布式計(jì)算框架，是它的一個(gè)基礎(chǔ)組件，當(dāng)配置好Hadoop集群時(shí)，MapReduce已包含在內(nèi)。下面先舉個(gè)簡(jiǎn)單的例子來(lái)幫助大家理解一下什么是MapReduce。5.1MapReduce簡(jiǎn)介如果某班級(jí)要組織一次春游活動(dòng)，班主任需要向?qū)W生收取春游的費(fèi)用，那么班主任會(huì)告訴班長(zhǎng)，讓他收取春游費(fèi)用，班長(zhǎng)把任務(wù)分給各組組長(zhǎng)，讓他們把各自組員的費(fèi)用收上來(lái)并交給他，最后班長(zhǎng)把收上來(lái)的錢(qián)交給老師。這就是一個(gè)典型的MapReduce過(guò)程，在這個(gè)例子中，班長(zhǎng)把任務(wù)分給各組組長(zhǎng)的過(guò)程稱(chēng)為Map過(guò)程；各組組長(zhǎng)把費(fèi)用收齊后再交給班長(zhǎng)進(jìn)行匯總的過(guò)程就是Reduce過(guò)程。5.1MapReduce簡(jiǎn)介MapReduce程序的第一步叫做mapping5.1MapReduce簡(jiǎn)介Reducing把數(shù)據(jù)聚集在一起5.1MapReduce簡(jiǎn)介簡(jiǎn)而言之，MapReduce采用的是一種“分而治之”的思想，即把一個(gè)大而重的任務(wù)拆解開(kāi)來(lái)，分成一系列小而輕的任務(wù)并行處理，這樣就使任務(wù)得以快速解決。5.1MapReduce簡(jiǎn)介從業(yè)界使用分布式系統(tǒng)的變化趨勢(shì)和Hadoop框架的長(zhǎng)遠(yuǎn)發(fā)展來(lái)看，MapReduce的JobTracker/TaskTracker機(jī)制需要大規(guī)模的調(diào)整來(lái)修復(fù)它在可擴(kuò)展性，內(nèi)存消耗，線程模型，可靠性和性能上的缺陷。在過(guò)去的幾年中，Hadoop開(kāi)發(fā)團(tuán)隊(duì)做了一些bug的修復(fù)，但是最近這些修復(fù)的成本越來(lái)越高，這表明對(duì)原框架做出改變的難度越來(lái)越大。5.1MapReduce簡(jiǎn)介為了從根本上解決舊MapReduce框架的性能瓶頸，促進(jìn)Hadoop框架的更長(zhǎng)遠(yuǎn)發(fā)展，從0.23.0版本開(kāi)始，Hadoop的MapReduce框架完全重構(gòu)，發(fā)生了根本的變化。新的HadoopMapReduce框架出現(xiàn)了，從初期的MapReduceV1(MRv1)到MapReduceV2（MRv2）,MRv2增加了YARN，YARN是資源管理和任務(wù)調(diào)度的框架，很好地減輕了MRv1的JobTracker的壓力。5.2MapReduce編程模型5.2MapReduce編程模型簡(jiǎn)單模型HadoopMapReduce編程模型主要由兩個(gè)抽象類(lèi)構(gòu)成，即Mapper和Reducer。Mapper用以對(duì)切分過(guò)的原始數(shù)據(jù)進(jìn)行處理，Reducer對(duì)Mapper的結(jié)果進(jìn)行匯總，得到最后的輸出。5.2MapReduce編程模型復(fù)雜模型對(duì)于大部分任務(wù)來(lái)說(shuō)，都是需要Reduce過(guò)程的，并且由于任務(wù)繁重，會(huì)啟動(dòng)多個(gè)Reducer（默認(rèn)為1，根據(jù)任務(wù)量可由用戶(hù)自己設(shè)定合適的Reducer數(shù)量）來(lái)進(jìn)行匯總，如果只用一個(gè)Reducer計(jì)算所有Mapper的結(jié)果，會(huì)導(dǎo)致單個(gè)Reducer負(fù)載過(guò)于繁重，形成性能瓶頸，大大增加任務(wù)的運(yùn)行周期。5.2MapReduce編程模型Key值歸并

5.3MapReduce數(shù)據(jù)流5.3MapReduce數(shù)據(jù)流高層MapReduce工作流水線

mapping任務(wù)平等，Mapper無(wú)特定標(biāo)識(shí)，處理任意輸入每個(gè)mapper本地進(jìn)行中間值交換，同鍵值送到同一個(gè)reducer。唯一的結(jié)點(diǎn)通信數(shù)據(jù)傳送由平臺(tái)處理5.3MapReduce數(shù)據(jù)流5.3MapReduce數(shù)據(jù)流輸入文件：任務(wù)的初始存儲(chǔ)地輸入格式：inputFormat定義了如何分割和讀取輸入文件。選擇對(duì)象劃分為inputSplits每個(gè)InputSplit一個(gè)map任務(wù)為RecordReader讀取文件提供一個(gè)工廠方法輸入格式描述鍵值TextInputFormat默認(rèn)格式，讀取文件的行行的字節(jié)偏移量行的內(nèi)容KeyValueInputFormat把行解析為鍵值對(duì)第一個(gè)tab字符前的所有字符行剩下的內(nèi)容SequenceFileInputFormatHadoop定義的高性能二進(jìn)制格式用戶(hù)自定義用戶(hù)自定義5.3MapReduce數(shù)據(jù)流輸入塊InputSplit：文件拆分成塊Mapred-default.xml設(shè)置塊大小輸入格式定義了組成mapping階段的map任務(wù)列表，每個(gè)任務(wù)對(duì)應(yīng)一個(gè)輸入塊。據(jù)InputSplit地址分配任務(wù)（最大任務(wù)數(shù)mapred.tasktracker.map.tasks.maximum）5.3MapReduce數(shù)據(jù)流RecordReader訪問(wèn)InputSplit轉(zhuǎn)換為適合mapper讀取的kv對(duì)RecordReader實(shí)例由輸入格式定義的默認(rèn)的輸入格式TextInputFormat提供LineRecordReader偏移量-鍵；行-值RecordReader會(huì)在輸入塊上被重復(fù)的調(diào)用直到整個(gè)輸入塊被處理完畢，每一次調(diào)用RecordReader都會(huì)調(diào)用Mapper的map()方法。5.3MapReduce數(shù)據(jù)流TextInputFormat演示5.3MapReduce數(shù)據(jù)流給定一個(gè)鍵值對(duì)，map()方法會(huì)生成一個(gè)或多個(gè)鍵值對(duì)OutputCollector對(duì)象collect()的方法，利用該方法把鍵值對(duì)送到作業(yè)的reduce階段。5.3MapReduce數(shù)據(jù)流Map過(guò)程

WordCountMapper5.3MapReduce數(shù)據(jù)流Shuffle:Shuffle過(guò)程是指Mapper產(chǎn)生的直接輸出結(jié)果，經(jīng)過(guò)一系列的處理，成為最終的Reducer直接輸入數(shù)據(jù)的整個(gè)過(guò)程Partitons:每一個(gè)reduce節(jié)點(diǎn)會(huì)分派到中間輸出的鍵集合中的一個(gè)不同的子集合Partitioner類(lèi)決定鍵值對(duì)去向，默認(rèn)分類(lèi)器計(jì)算鍵哈希值5.3MapReduce數(shù)據(jù)流Mapper端shuffle–spill操作5.3MapReduce數(shù)據(jù)流Partiton&ShuffleMapper端shuffle–partition操作5.3MapReduce數(shù)據(jù)流Mapper端shuffle演示（本演示內(nèi)容很少，只有一個(gè)Reducer，沒(méi)必要進(jìn)行Partition操作）5.3MapReduce數(shù)據(jù)流Reducer端shuffle5.3MapReduce數(shù)據(jù)流Reducer端shuffle演示（如果設(shè)置了combine，在mergesort時(shí)會(huì)進(jìn)行相應(yīng)操作）5.3MapReduce數(shù)據(jù)流每個(gè)reduce任務(wù)都會(huì)創(chuàng)建一個(gè)Reducer實(shí)例reducer的reduce()方法只會(huì)調(diào)用一次，它會(huì)接收一個(gè)鍵和關(guān)聯(lián)到鍵的所有值的一個(gè)迭代器，迭代器會(huì)以一個(gè)未定義的順序返回關(guān)聯(lián)到同一個(gè)鍵的值。5.3MapReduce數(shù)據(jù)流WordCountReducer5.3MapReduce數(shù)據(jù)流輸出格式描述TextOutputFormat默認(rèn)的輸出格式，以"key\tvalue"的方式輸出行SequenceFileOutputFormat輸出二進(jìn)制文件，適合于讀取為子MapReduce作業(yè)的輸入NullOutputFormat忽略收到的數(shù)據(jù)，即不做輸出

鍵值對(duì)通過(guò)OutputCollector被寫(xiě)到輸出文件中，寫(xiě)入的方式由輸出格式控制。每一個(gè)reducer會(huì)把結(jié)果輸出寫(xiě)在公共文件夾中一個(gè)單獨(dú)的文件內(nèi)，這些文件的命名一般是part-nnnnn，nnnnn是關(guān)聯(lián)到某個(gè)reduce任務(wù)的partition的id5.3MapReduce數(shù)據(jù)流TextOutputFormat演示5.4MapReduce任務(wù)流程5.4MapReduce任務(wù)流程整體流程圖MRV2MRV2+YARN5.4MapReduce任務(wù)流程客戶(hù)端用于向Yarn集群提交任務(wù)MRAppMaster為ApplicationMaster的一個(gè)實(shí)現(xiàn)，它監(jiān)控和調(diào)度一整套MR任務(wù)流程，每個(gè)MR任務(wù)只產(chǎn)生一個(gè)MRAppMaster。MRAppMaster只負(fù)責(zé)任務(wù)管理，并不負(fù)責(zé)資源的調(diào)配。用戶(hù)定義的Map函數(shù)和Reduce函數(shù)的實(shí)例化，在MRv2中，它們只能運(yùn)行在Yarn給定的資源限制下，由MRAppMaster和NodeManage協(xié)同管理和調(diào)度。MRv2基本組成：5.4MapReduce任務(wù)流程Yarn基本組成：RM：為整個(gè)集群的資源調(diào)度器ResourceSchedule：當(dāng)有應(yīng)用程序已經(jīng)注冊(cè)需要運(yùn)行時(shí)，ApplicationMaster會(huì)向它申請(qǐng)資源，而它會(huì)根據(jù)當(dāng)時(shí)的資源和限制進(jìn)行資源分配，它會(huì)產(chǎn)生一個(gè)container資源描述。ApplicationsManager：它負(fù)責(zé)管理整個(gè)集群運(yùn)行的所有任務(wù)，包括應(yīng)用程序的提交、ResourceSchedule協(xié)商啟動(dòng)和監(jiān)控ApplicationMaster，并在ApplicationMaster任務(wù)失敗時(shí)在其他節(jié)點(diǎn)重啟它5.4MapReduce任務(wù)流程Yarn基本組成：負(fù)責(zé)對(duì)一個(gè)任務(wù)流程的調(diào)度、管理，包括任務(wù)注冊(cè)、資源申請(qǐng)以及和NodeManage通信以開(kāi)啟和殺死任務(wù)等。運(yùn)行于datanode，監(jiān)控并管理單個(gè)節(jié)點(diǎn)的計(jì)算資源，并定時(shí)向RM匯報(bào)節(jié)點(diǎn)的資源使用情況，當(dāng)節(jié)點(diǎn)上有任務(wù)時(shí)，還負(fù)責(zé)對(duì)container進(jìn)行創(chuàng)建、運(yùn)行狀態(tài)的監(jiān)控及最終銷(xiāo)毀。Yarn架構(gòu)下對(duì)運(yùn)算資源的一種描述，它封裝了某個(gè)節(jié)點(diǎn)的多維度資源，包括cpu、ram、disk、network等。當(dāng)AM向RM申請(qǐng)資源時(shí)，RM分配的資源就是以container表示的，Maptask和ReduceTask只能在所分配的container描述限制中運(yùn)行。5.4MapReduce任務(wù)流程client向ResourceManager提交任務(wù)ResourceManager分配該任務(wù)的第一個(gè)container，并通知相應(yīng)的NodeManager啟動(dòng)MRAppMaster。5.4MapReduce任務(wù)流程N(yùn)odeManager接受命令后，開(kāi)辟一個(gè)container資源空間，并在container中啟動(dòng)相應(yīng)的MRAppMaster。5.4MapReduce任務(wù)流程MRAppMaster啟動(dòng)之后，第一步會(huì)向ResourceManager注冊(cè)，這樣用戶(hù)可以直接通過(guò)MRAppMaster監(jiān)控任務(wù)的運(yùn)行狀態(tài)；之后則直接由MRAppMaster調(diào)度任務(wù)運(yùn)行，重復(fù)5~8，直到任務(wù)結(jié)束。5.4MapReduce任務(wù)流程MRAppMaster以輪詢(xún)的方式向ResourceManager申請(qǐng)任務(wù)運(yùn)行所需的資源。5.4MapReduce任務(wù)流程一旦ResourceManager配給了資源，MRAppMaster便會(huì)與相應(yīng)的NodeManager通信，讓它劃分Container并啟動(dòng)相應(yīng)的任務(wù)（MapTask或ReduceTask）。5.4MapReduce任務(wù)流程N(yùn)odeManager準(zhǔn)備好運(yùn)行環(huán)境，啟動(dòng)任務(wù)。5.4MapReduce任務(wù)流程各任務(wù)運(yùn)行，并定時(shí)通過(guò)RPC協(xié)議向MRAppMaster匯報(bào)自己的運(yùn)行狀態(tài)和進(jìn)度。MRAppMaster也會(huì)實(shí)時(shí)地監(jiān)控任務(wù)的運(yùn)行，當(dāng)發(fā)現(xiàn)某個(gè)Task假死或失敗時(shí)，便殺死它重新啟動(dòng)任務(wù)。5.4MapReduce任務(wù)流程任務(wù)完成，MRAppMaster向ResourceManager通信，注銷(xiāo)并關(guān)閉自己。5.5MapReduce的Streaming和Pipe5.5MapReduce的Streaming和PipeHadoopStreamingHadoopStreaming可以將任何可執(zhí)行的腳本或二進(jìn)制文件封裝成Mapper或Reducer，可以大大提高M(jìn)apReduce程序的開(kāi)發(fā)效率。Streaming啟動(dòng)的MR作業(yè)使用標(biāo)準(zhǔn)輸入輸出與用戶(hù)的MapReduce進(jìn)行數(shù)據(jù)傳遞，所以要求用戶(hù)編寫(xiě)的程序必須以標(biāo)準(zhǔn)輸入作為數(shù)據(jù)入口，標(biāo)準(zhǔn)輸出作為數(shù)據(jù)出口。使用Streaming時(shí)，用戶(hù)需要提供兩個(gè)可執(zhí)行文件，一個(gè)用于Mappper，一個(gè)用于Reducer，當(dāng)一個(gè)可執(zhí)行文件被用于Mapper或Reducer時(shí)，在初始化時(shí)，它們會(huì)作為一個(gè)單獨(dú)的進(jìn)程啟動(dòng)，而Mappper和Reducer則充當(dāng)封裝傳遞角色，把輸入切分成行，傳給相應(yīng)的可執(zhí)行文件處理，同時(shí)收集由可執(zhí)行文件的標(biāo)準(zhǔn)輸出，并轉(zhuǎn)化為<key,value>的形式作為相應(yīng)的Mappper和Reducer輸出，Reducer的輸出會(huì)直接寫(xiě)入HDFS。5.5MapReduce的Streaming和PipeHadoopPipePipe是專(zhuān)為C/C++用戶(hù)設(shè)計(jì)的MapReduce編程工具，它的設(shè)計(jì)思想是把相關(guān)C/C++代碼（包括Map和Reduce）封裝在一個(gè)單獨(dú)的進(jìn)程中，運(yùn)行時(shí)通過(guò)套接字（Socket）與Java端進(jìn)行數(shù)據(jù)傳遞。5.6MapReduce的性能調(diào)優(yōu)5.6MapReduce的性能調(diào)優(yōu)

在執(zhí)行MapReduce任務(wù)時(shí)，由于數(shù)據(jù)的差異性，使用默認(rèn)的置往往并不能完全發(fā)揮集群的運(yùn)算性能，所以需要對(duì)參數(shù)進(jìn)行相應(yīng)的調(diào)整或者適當(dāng)預(yù)處理，以?xún)?yōu)化任務(wù)的運(yùn)行效率。但是不當(dāng)?shù)恼{(diào)整反而會(huì)影響任務(wù)的執(zhí)行，所以用戶(hù)必須對(duì)自己的集群和要處理的數(shù)據(jù)比較熟悉，這樣進(jìn)行的配置才會(huì)有價(jià)值。但大部分情況，進(jìn)行某一項(xiàng)優(yōu)化之后，往往會(huì)影響另一項(xiàng)執(zhí)行流程，如何在盡量不影響其他執(zhí)行流程的情況下優(yōu)化某一配置，這是用戶(hù)需要慎重考慮的問(wèn)題，所以這是一個(gè)權(quán)衡與妥協(xié)的過(guò)程，并沒(méi)有所謂的最優(yōu)解。5.6MapReduce的性能調(diào)優(yōu)1．合并小文件和合理設(shè)置Mapper數(shù)量2．推測(cè)執(zhí)行3．優(yōu)化每個(gè)節(jié)點(diǎn)同時(shí)運(yùn)行的任務(wù)數(shù)4．合理設(shè)置Reducer數(shù)量5．壓縮Map輸出6．JVM重用7．JVM內(nèi)存本章完第六章HadoopIO操作6.1HDFS數(shù)據(jù)完整性6.1HDFS數(shù)據(jù)完整性

由于每個(gè)磁盤(pán)或者網(wǎng)絡(luò)上的I/O操作可能會(huì)對(duì)正在讀寫(xiě)的數(shù)據(jù)不慎引入錯(cuò)誤，如果通過(guò)的數(shù)據(jù)流量非常大，數(shù)據(jù)發(fā)生損壞的幾率很高。

檢查損壞數(shù)據(jù)的常用方法是在第一次進(jìn)入系統(tǒng)時(shí)計(jì)算數(shù)據(jù)的校驗(yàn)和，然后只要數(shù)據(jù)不是在一個(gè)可靠的通道上傳輸，就可能會(huì)發(fā)生損壞。如果新生成的校驗(yàn)和不完全匹配原始的校驗(yàn)和，那么數(shù)據(jù)就會(huì)被認(rèn)為是損壞的。

一個(gè)常用的錯(cuò)誤檢測(cè)代碼是CRC-32(cyclicredundancycheck,循環(huán)冗余檢查)，計(jì)算一個(gè)32位的任何大小輸入的整數(shù)校驗(yàn)和。6.1HDFS數(shù)據(jù)完整性HDFS以透明方式校驗(yàn)所有寫(xiě)入它的數(shù)據(jù)，并在默認(rèn)設(shè)置下，會(huì)在讀取數(shù)據(jù)時(shí)驗(yàn)證校驗(yàn)和。針對(duì)數(shù)據(jù)的每個(gè)io.bytes.per.checksum(默認(rèn)512字節(jié))字節(jié)，都會(huì)創(chuàng)建一個(gè)單獨(dú)的校驗(yàn)和。數(shù)據(jù)節(jié)點(diǎn)負(fù)責(zé)在存儲(chǔ)數(shù)據(jù)及其校驗(yàn)和之前驗(yàn)證它們收到的數(shù)據(jù)。從客戶(hù)端和其它數(shù)據(jù)節(jié)點(diǎn)復(fù)制過(guò)來(lái)的數(shù)據(jù)?？蛻?hù)端寫(xiě)入數(shù)據(jù)并且將它發(fā)送到一個(gè)數(shù)據(jù)節(jié)點(diǎn)管線中，在管線的最后一個(gè)數(shù)據(jù)節(jié)點(diǎn)驗(yàn)證校驗(yàn)和。6.1HDFS數(shù)據(jù)完整性客戶(hù)端讀取數(shù)據(jù)節(jié)點(diǎn)上的數(shù)據(jù)時(shí)，會(huì)驗(yàn)證校驗(yàn)和，將其與數(shù)據(jù)節(jié)點(diǎn)上存儲(chǔ)的校驗(yàn)和進(jìn)行對(duì)比。每個(gè)數(shù)據(jù)節(jié)點(diǎn)維護(hù)一個(gè)連續(xù)的校驗(yàn)和驗(yàn)證日志，因此它知道每個(gè)數(shù)據(jù)塊最后驗(yàn)證的時(shí)間。每個(gè)數(shù)據(jù)節(jié)點(diǎn)還會(huì)在后臺(tái)線程運(yùn)行一個(gè)DataBlockScanner（數(shù)據(jù)塊檢測(cè)程序），定期驗(yàn)證存儲(chǔ)在數(shù)據(jù)節(jié)點(diǎn)上的所有塊，為了防止物理存儲(chǔ)介質(zhì)中位衰減鎖造成的數(shù)據(jù)損壞6.1HDFS數(shù)據(jù)完整性6.1HDFS數(shù)據(jù)完整性6.1HDFS數(shù)據(jù)完整性HDFS通過(guò)復(fù)制完整的副本來(lái)產(chǎn)生一個(gè)新的，無(wú)錯(cuò)的副本來(lái)“治愈”哪些出錯(cuò)的數(shù)據(jù)塊。工作方式：如果客戶(hù)端讀取數(shù)據(jù)塊時(shí)檢測(cè)到錯(cuò)誤，拋出ChecksumException前報(bào)告該壞塊以及它試圖從名稱(chēng)節(jié)點(diǎn)中藥讀取的數(shù)據(jù)節(jié)點(diǎn)。名稱(chēng)節(jié)點(diǎn)將這個(gè)塊標(biāo)記為損壞的，不會(huì)直接復(fù)制給客戶(hù)端或復(fù)制該副本到另一個(gè)數(shù)據(jù)節(jié)點(diǎn)。它會(huì)從其他副本復(fù)制一個(gè)新的副本。6.1HDFS數(shù)據(jù)完整性本地文件系統(tǒng)也可能禁用校驗(yàn)和：底層文件系統(tǒng)原生支持校驗(yàn)和。這里通過(guò)RawLocalFileSystem來(lái)替代LocalFileSystem完成。要在一個(gè)應(yīng)用中全局使用，只需要設(shè)置fs.file.impl值為org.apache.hadoop.fs.RawLocalFileSystem來(lái)重新map執(zhí)行文件的URL?；蛘咧幌雽?duì)某些讀取禁用校驗(yàn)和校驗(yàn)。

Configurationconf=...

FileSystemfs=newRawLocalFileSystem();

fs.initialize(null,conf);6.1HDFS數(shù)據(jù)完整性ChecksumFileSystemLocalFileSystem使用ChecksumFileSystem(校驗(yàn)和文件系統(tǒng))為自己工作，這個(gè)類(lèi)可以很容易添加校驗(yàn)和功能到其他文件系統(tǒng)中。因?yàn)镃hecksumFileSystem也包含于文件系統(tǒng)中。

FileSystemrawFs=...

FileSystemchecksummedFs=newChecksumFileSystem(rawFs);6.2基于文件的數(shù)據(jù)結(jié)構(gòu)6.2基于文件的數(shù)據(jù)結(jié)構(gòu)Hadoop的HDFS和MapReduce子框架主要是針對(duì)大數(shù)據(jù)文件來(lái)設(shè)計(jì)的，在小文件的處理上不但效率低下，而且十分消耗內(nèi)存資源。解決辦法通常是選擇一個(gè)容器，將這些小文件包裝起來(lái)，將整個(gè)文件作為一條記錄，可以獲得更高效率的存儲(chǔ)和處理，避免了多次打開(kāi)關(guān)閉流耗費(fèi)計(jì)算資源。HDFS提供了兩種類(lèi)型的容器，分別是SequenceFile和MapFile。6.2基于文件的數(shù)據(jù)結(jié)構(gòu)SequenceFile存儲(chǔ)SequenceFile的存儲(chǔ)類(lèi)似于日志文件，所不同的是日志文件的每條記錄都是純文本數(shù)據(jù)，而SequenceFile的每條記錄是可序列化、可持久化的鍵值數(shù)據(jù)結(jié)構(gòu)。SequenceFile提供相應(yīng)的讀寫(xiě)器和排序器，寫(xiě)操作根據(jù)壓縮的類(lèi)型分為3種。（1）Writer：無(wú)壓縮寫(xiě)數(shù)據(jù)。（2）RecordCompressWriter：記錄級(jí)壓縮文件，只壓縮值。（3）BlockCompressWrite：塊級(jí)壓縮文件，鍵值采用獨(dú)立壓縮方式。讀取操作實(shí)際上可以讀取上述3種類(lèi)型。6.2基于文件的數(shù)據(jù)結(jié)構(gòu)SequenceFile存儲(chǔ)在存儲(chǔ)結(jié)構(gòu)上，SequenceFile主要由一個(gè)Header后跟多條Record組成，如圖所示。6.2基于文件的數(shù)據(jù)結(jié)構(gòu)SequenceFile存儲(chǔ)當(dāng)保存的記錄有很多的時(shí)候，可以把一連串的記錄組織到一起，統(tǒng)一壓縮成一個(gè)塊。6.2基于文件的數(shù)據(jù)結(jié)構(gòu)用命令行接口顯示序列文件使用-text選項(xiàng)顯示文本格式的序列文件。

%hadoopfs-textnumber.seq6.2基于文件的數(shù)據(jù)結(jié)構(gòu)MapFile是排序后的SequenceFile，并且它會(huì)額外生成一個(gè)索引文件提供按鍵的查找。讀寫(xiě)MapFile與讀寫(xiě)SequenceFile非常類(lèi)似，只需要換成MapFie.Reader和MapFile.Writer就可以了。在命令行顯示MapFile的文件內(nèi)容同樣要用-text。MapFile

6.3壓縮6.3壓縮在執(zhí)行MapReduce程序的過(guò)程中，在Mapper端和Reducer端會(huì)進(jìn)行大量的數(shù)據(jù)傳輸和磁盤(pán)的I/O操作，在這個(gè)過(guò)程中對(duì)數(shù)據(jù)進(jìn)行壓縮處理，可以有效減少底層存儲(chǔ)（HDFS）讀寫(xiě)的字節(jié)數(shù)，并且通過(guò)減少M(fèi)ap和Reduce階段數(shù)據(jù)的I/O來(lái)提升MapReduce程序的運(yùn)行速度，提高網(wǎng)絡(luò)帶寬，節(jié)約磁盤(pán)空間。在Hadoop下，尤其是在數(shù)據(jù)規(guī)模很大和工作負(fù)載密集的情況下，對(duì)數(shù)據(jù)進(jìn)行壓縮處理非常重要。6.3壓縮文件壓縮兩大好處：減少存儲(chǔ)文件所需要的空間且加快了數(shù)據(jù)在網(wǎng)絡(luò)上或從磁盤(pán)上或到磁盤(pán)上的傳輸速度。各種壓縮算法的壓縮比：

6.3壓縮編碼和解碼Codec是Coder與Decoder的縮略詞，Codec類(lèi)用于實(shí)現(xiàn)一種壓縮-解壓算法。Hadoop中的壓縮與解壓的類(lèi)是在CompressionCodec接口下實(shí)現(xiàn)的。我們這里講的Codec就是實(shí)現(xiàn)了CompressionCodec接口的一些壓縮格式的類(lèi)。6.3壓縮編碼和解碼CompressionCodec有兩個(gè)方法輕松地壓縮和解壓數(shù)據(jù)。使用usethe

createOutputStream(OutputStreamout)創(chuàng)建一個(gè)CompressionOutputStream，將其以壓縮格式寫(xiě)入底層的流。使用createInputStream(InputStreamin)獲取一個(gè)CompressionInputStream，從底層的流讀取未壓縮的數(shù)據(jù)。

6.3壓縮編碼和解碼01packagecom.laos.hadoop;0203importorg.apache.hadoop.conf.Configuration;04importorg.apache.hadoop.io.IOUtils;05importpress.CompressionCodec;06importpress.CompressionOutputStream;07importorg.apache.hadoop.util.ReflectionUtils;0809publicclassStreamCompressor{10

publicstaticvoidmain(String[]args)throwsException{11

StringcodecClassname="press.GzipCodec";12

Class<?>codecClass=Class.forName(codecClassname);13

Configurationconf=newConfiguration();14

CompressionCodeccodec=(CompressionCodec)ReflectionUtils15

.newInstance(codecClass,conf);16

//將讀入數(shù)據(jù)壓縮至System.out17

CompressionOutputStreamout=codec.createOutputStream(System.out);18

IOUtils.copyBytes(System.in,out,4096,false);19

out.finish();20

}2122}

在unix窗口輸入命令：$echo"Test"|hadoopjarhadoop-itest.jarcom.laos.hadoop.StreamCompressor|gunzip6.3壓縮編碼和解碼

01packagecom.laos.hadoop;0203importjava.io.InputStream;04importjava.io.OutputStream;05import.URI;0607importorg.apache.hadoop.conf.Configuration;08importorg.apache.hadoop.fs.FileSystem;09importorg.apache.hadoop.fs.Path;10importorg.apache.hadoop.io.IOUtils;11importpress.CompressionCodec;12importpress.CompressionCodecFactory;1314publicclassFileDecompressor{15

publicstaticvoidmain(String[]args)throwsException{16

Stringuri=args[0];17

Configurationconf=newConfiguration();18

FileSystemfs=FileSystem.get(URI.create(uri),conf);1920

PathinputPath=newPath(uri);21

CompressionCodecFactoryfactory=newCompressionCodecFactory(conf);22

CompressionCodeccodec=factory.getCodec(inputPath);23

if(codec==null){24

System.err.println("Nocodecfoundfor"+uri);25

System.exit(1);26

}27

StringoutputUri=CompressionCodecFactory.removeSuffix(uri,codec28

.getDefaultExtension());29

InputStreamin=null;30

OutputStreamout=null;31

try{32

in=codec.createInputStream(fs.open(inputPath));33

out=fs.create(newPath(outputUri));34

IOUtils.copyBytes(in,out,conf);35

}finally{36

IOUtils.closeStream(in);37

IOUtils.closeStream(out);38

}39

}40}6.3壓縮壓縮和輸入分隔考慮如何壓縮哪些將由MapReduce處理的數(shù)據(jù)時(shí)，考慮壓縮格式是否支持分隔很重要。

例如，gzip格式使用default來(lái)存儲(chǔ)壓縮過(guò)的數(shù)據(jù)，default將數(shù)據(jù)作為一系列壓縮過(guò)的塊存儲(chǔ)，但是每塊的開(kāi)始沒(méi)有指定用戶(hù)在數(shù)據(jù)流中的任意點(diǎn)定位到下一個(gè)塊的起始位置，而是自身與數(shù)據(jù)同步，所以gzip不支持分隔機(jī)制。6.3壓縮在MapReduce中使用壓縮如果要壓縮MapReduce作業(yè)的輸出，設(shè)置press為true，pression.codec屬性指定編碼解碼器。

如果輸入的文件時(shí)壓縮過(guò)的，MapReduce讀取時(shí)，它們會(huì)自動(dòng)解壓，根據(jù)文件擴(kuò)展名來(lái)決定使用那一個(gè)壓縮解碼器。publicclassMaxTemperatureWithCompression{17

publicstaticvoidmain(String[]args)throwsIOException{18

if(args.length!=2){19

System.err.println("Usage:MaxTemperatureWithCompression<inputpath>"+20

"<outputpath>");21

System.exit(-1);22

}23

24

JobConfconf=newJobConf(MaxTemperatureWithCompression.class);conf.setJobName("Maxtemperaturewithoutputcompression");25

FileInputFormat.addInputPath(conf,newPath(args[0]));26

FileOutputFormat.setOutputPath(conf,newPath(args[1]));27

28

conf.setOutputKeyClass(Text.class);29

conf.setOutputValueClass(IntWritable.class);30

31

conf.setBoolean("press",true);32

conf.setClass("pression.codec",GzipCodec.class,33

CompressionCodec.class);

34

JobClient.runJob(conf);35

}36

}6.4序列化6.4序列化序列化：將結(jié)構(gòu)化對(duì)象轉(zhuǎn)換為字節(jié)流以便于通過(guò)網(wǎng)絡(luò)進(jìn)行傳輸或?qū)?/p>