MapReduce基礎(chǔ)編程mapreduce基礎(chǔ)編程

Hadoop大數(shù)據(jù)處理實戰(zhàn)第6章MapReduce基礎(chǔ)與編程本章導(dǎo)讀MapReduce是Hadoop中的大數(shù)據(jù)并行處理框架，其原理簡單且易于實現(xiàn)。它將一個數(shù)據(jù)處理任務(wù)劃分成Map和Reduce兩個任務(wù)，使得用戶可以在不清楚分布式計算框架內(nèi)部運行機(jī)制的情況下輕松完成計算任務(wù)。本章首先介紹分布式并行計算的概念及MapReduce的含義和特點，然后介紹MapReduce的工作流程，并重點介紹Shuffle過程，最后通過單詞統(tǒng)計和二次排序兩個案例具體介紹如何利用MapReduce進(jìn)行編程。學(xué)習(xí)目標(biāo)020304了解分布式并行計算的概念。掌握MapReduce的基礎(chǔ)知識。熟悉MapReduce的工作流程。掌握MapReduce的Shuffle過程。掌握MapReduce的編程方法。01MapReduce概述MapReduce的工作流程02目錄CONTENTS案例解析——單詞統(tǒng)計編程03案例解析——二次排序編程0401MapReduce概述MapReduce是Hadoop系統(tǒng)中最重要的計算引擎，它不僅直接支持交互式應(yīng)用、基于程序的應(yīng)用，還是Hive等組件的基礎(chǔ)。MapReduce概述MapReduce概述6.1.1分布式并行計算分布式計算（distributedcomputing）可以將大任務(wù)拆分成許多小任務(wù)，然后把這些任務(wù)派發(fā)給多臺計算機(jī)進(jìn)行計算，最后將所有的計算結(jié)果進(jìn)行匯總，以得到最終的結(jié)果。此外，分解后的任務(wù)之間互相獨立，計算結(jié)果之間幾乎互不影響，且實時性要求不高。1．分布式計算并行計算（parallelcomputing）又稱平行計算，它是一種允許讓多條指令以平行的方式同時進(jìn)行計算的模式，包括時間并行和空間并行兩種方式。其中，時間并行可理解為利用多條流水線同時作業(yè)，而空間并行可認(rèn)為是使用多種計算資源（如多個處理器）執(zhí)行并發(fā)計算，從而減少解決復(fù)雜問題所花費的時間。2．并行計算對于大多數(shù)開發(fā)人員而言，并行計算是復(fù)雜而陌生的，至于分布式計算，直接進(jìn)行編程就更加困難了。而MapReduce的出現(xiàn)，很好地解決了分布式并行計算的難題。MapReduce概述6.1.2MapReduce的含義受Google發(fā)表的MapReduce論文的核心思想啟發(fā)，HadoopMapReduce將Google的核心計算模型MapReduce進(jìn)行了開源實現(xiàn)，使之可以在大規(guī)模集群上并行處理海量數(shù)據(jù)。通俗地講，MapReduce是一個針對大規(guī)模集群中的分布式文件進(jìn)行并行處理的計算模型。MapReduce原理簡單且易于實現(xiàn)，其設(shè)計目標(biāo)就是讓不熟悉分布式并行編程的開發(fā)人員，將自己的程序輕松運行在分布式系統(tǒng)上。此外，MapReduce屏蔽了分布式并行計算的諸多細(xì)節(jié)，僅為用戶提供MapReduce計算接口，從而使得開發(fā)人員只需進(jìn)行簡單的編程就可以編寫并行計算程序。MapReduce概述MapReduce主要將數(shù)據(jù)處理任務(wù)拆分為Map（映射）和Reduce（規(guī)約）兩個任務(wù)。MapReduce充分體現(xiàn)了“分而治之”的思想，即把一個復(fù)雜的任務(wù)拆分成小的任務(wù)并行處理，從而提高了任務(wù)的處理速度。某個房屋銷售公司想要統(tǒng)計某月的銷售總額，老板就會讓總經(jīng)理報告銷售總額，而總經(jīng)理則會把統(tǒng)計任務(wù)分配給各地區(qū)經(jīng)理，讓他們各自統(tǒng)計其管轄區(qū)域的銷售總額，最后總經(jīng)理將各地區(qū)的銷售總額匯總后報告給老板。這就是一個典型的MapReduce任務(wù)：總經(jīng)理將銷售額統(tǒng)計任務(wù)分給各地區(qū)經(jīng)理，相當(dāng)于Map任務(wù)，而總經(jīng)理匯總各地區(qū)的銷售額，就是Reduce任務(wù)。MapReduce概述6.1.3MapReduce的特點MapReduce作為Hadoop的核心組件之一，它在數(shù)據(jù)處理方面具有自己的特色。概括而言，MapReduce在大數(shù)據(jù)處理方面具有以下幾個優(yōu)勢：

1（1）開發(fā)簡單且易于實現(xiàn)。MapReduce在執(zhí)行分布式并行計算時，會將并行編程的繁瑣細(xì)節(jié)隱藏起來，如任務(wù)調(diào)度、負(fù)載均衡、失敗恢復(fù)等。所以，開發(fā)人員在設(shè)計程序時，只需要設(shè)計簡單的計算邏輯，就可以完成分布式計算任務(wù)，這就大大降低了分布式程序的編寫難度。MapReduce概述

2（2）良好的可擴(kuò)展性。MapReduce具有良好的可擴(kuò)展性，這意味著當(dāng)集群計算資源不足時，可以通過動態(tài)增加節(jié)點的方式實現(xiàn)彈性計算。

3如果集群中的某計算節(jié)點出現(xiàn)故障，使得作業(yè)執(zhí)行失敗，MapReduce可自動將作業(yè)分配到可用的計算節(jié)點上重新執(zhí)行。（3）高容錯性。MapReduce概述MapReduce也存在以下局限性：（1）計算局限。MapReduce適用于海量數(shù)據(jù)的離線批處理，但不能像Oracle數(shù)據(jù)庫那樣在毫秒時間內(nèi)返回計算結(jié)果。因此，它不適合數(shù)據(jù)事務(wù)處理或單一請求處理。如果用戶需要大數(shù)據(jù)的毫秒級響應(yīng)，可以考慮HBase應(yīng)用程序。此外，MapReduce中的輸入數(shù)據(jù)是靜態(tài)的，所以它也不適合流式數(shù)據(jù)計算。如果用戶想要進(jìn)行流式數(shù)據(jù)計算，可以使用Storm、SparkStreaming和Flink等流式計算框架，并且它們同MapReduce一樣，都可以在YARN上運行。（2）性能局限。Map任務(wù)與Reduce任務(wù)存在嚴(yán)格的依賴關(guān)系。Map任務(wù)的中間結(jié)果存儲在本地磁盤上，然后Reduce任務(wù)需要從磁盤上獲取Map計算的中間結(jié)果并將其作為輸入。這樣一來，就會產(chǎn)生大量的磁盤I/O，從而使得計算性能降低。不過，Spark與MapReduce不同，它在內(nèi)存中運算，這樣就避免了磁盤I/O。（3）應(yīng)用局限。MapReduce不適合一般的Web應(yīng)用，因為這些應(yīng)用只是簡單的數(shù)據(jù)訪問且每次訪問請求所需要的資源非常少，同時還要滿足高并發(fā)訪問需求。02MapReduce的工作流程對于某些簡單的數(shù)據(jù)處理任務(wù)，如文件中的數(shù)據(jù)只做簡單的數(shù)據(jù)格式轉(zhuǎn)換或切分等，可能只需要執(zhí)行Map任務(wù)就夠了，然后將Map任務(wù)輸出的結(jié)果直接存儲到HDFS中。但對于大多數(shù)復(fù)雜的計算任務(wù)來說，往往離不開Reduce任務(wù)。用戶需根據(jù)實際的業(yè)務(wù)需求設(shè)置合適的Reduce任務(wù)數(shù)（默認(rèn)為1），來完成數(shù)據(jù)處理任務(wù)。MapReduce的工作流程可將MapReduce的工作流程分為5個階段輸入分片和數(shù)據(jù)格式化Map過程Shuffle過程Reduce過程結(jié)果輸出MapReduce的工作流程在MapReduce計算過程中，以<key,value>鍵值對的形式進(jìn)行數(shù)據(jù)傳遞或計算，MapReduce的簡易數(shù)據(jù)處理流程如下圖所示。MapReduce的簡易數(shù)據(jù)處理流程MapReduce的工作流程MapReduce的工作流程6.2.1輸入分片和數(shù)據(jù)格式化在執(zhí)行Map任務(wù)之前，MapReduce會將存儲在分布式文件系統(tǒng)中的大規(guī)模數(shù)據(jù)集切分成獨立的輸入分片（InputSplit），并且每一個輸入分片對應(yīng)著一個Map任務(wù)。也就是說，有多少個輸入分片就會存在多少個Map任務(wù)。輸入分片是一個邏輯概念，它對輸入數(shù)據(jù)集的切分不是物理意義上的切分，而是對數(shù)據(jù)的邏輯結(jié)構(gòu)進(jìn)行切分。每個輸入分片存儲的并不是真實數(shù)據(jù)，而是指向分片數(shù)據(jù)的引用。例如，輸入分片中存儲了一些元數(shù)據(jù)信息，包括起始位置、數(shù)據(jù)長度、數(shù)據(jù)所在節(jié)點等。此外，由于Map任務(wù)的輸入數(shù)據(jù)要求是鍵值對的形式，所以需要對輸入分片進(jìn)行格式化，即將輸入分片處理成<key1,value1>形式的數(shù)據(jù)，然后再傳遞給Map任務(wù)。MapReduce的工作流程6.2.2Map過程Map過程利用map()函數(shù)來處理數(shù)據(jù)，map()函數(shù)接收<key1,value1>形式的數(shù)據(jù)輸入。經(jīng)過map()函數(shù)處理后，獲得一系列<key2,value2>形式的輸出。其中，map()函數(shù)中具體的數(shù)據(jù)處理方法可以由用戶自己定義。

map()函數(shù)的調(diào)用MapReduce的工作流程6.2.3Shuffle過程通常，map()函數(shù)的輸出并不會直接交給Reduce任務(wù)，而是需要經(jīng)過一系列處理，如分區(qū)（Partition）、排序（Sort）、合并（Combine）、溢寫（Spill，將內(nèi)存中的數(shù)據(jù)寫入磁盤）、歸并（Merge）等，然后將處理后的數(shù)據(jù)作為Reduce任務(wù)的輸入。這一系列處理過程稱為Shuffle（洗牌）過程。

MapReduce的Shuffle過程MapReduce的工作流程（1）map()函數(shù)的輸出并不會立即寫入磁盤，MapReduce會為每個Map任務(wù)分配一個環(huán)形內(nèi)存緩沖區(qū)（bufferinmemory），用于存儲map()函數(shù)的輸出。（2）在將環(huán)形內(nèi)存緩沖區(qū)中的數(shù)據(jù)寫入磁盤之前，需要對數(shù)據(jù)進(jìn)行分區(qū)、排序和合并（可選）操作。①分區(qū)操作的主要目的是將數(shù)據(jù)均勻地分配給Reduce任務(wù)，以實現(xiàn)MapReduce的負(fù)載均衡，從而避免單個Reduce任務(wù)的壓力過大。②排序操作是MapReduce的默認(rèn)操作，主要是將Map任務(wù)的輸出按key進(jìn)行排序。排序操作在分區(qū)操作之后執(zhí)行，因此，每個分區(qū)中的數(shù)據(jù)都是有序的。③排序結(jié)束后，用戶可根據(jù)實際需求選擇是否要執(zhí)行合并操作。不過，只有預(yù)先定義了Combine()函數(shù)，才會執(zhí)行合并操作，從而減少溢寫的數(shù)據(jù)量。所謂合并操作，就是將具有相同key的<key,value>的value加起來。例如，具有相同鍵的<'a',1>和<'a',1>兩個鍵值對，經(jīng)過合并操作之后，得到的結(jié)果為<'a',2>，這樣一來，鍵值對的數(shù)量就減少了。1．Map端的Shuffle過程MapReduce的工作流程（3）環(huán)形內(nèi)存緩沖區(qū)中的數(shù)據(jù)一旦達(dá)到閾值，一個后臺線程便開始把數(shù)據(jù)溢寫到本地磁盤的臨時文件（即溢寫文件）中。在溢寫到磁盤的過程中，Map的輸出仍然不斷地寫到環(huán)形內(nèi)存緩沖區(qū)中。不過，當(dāng)整個環(huán)形內(nèi)存緩沖區(qū)被數(shù)據(jù)占滿時，Map任務(wù)就會被阻塞，直到寫磁盤過程完成，才可以向環(huán)形內(nèi)存緩沖區(qū)繼續(xù)寫數(shù)據(jù)。（4）由于此時Map任務(wù)并未結(jié)束，系統(tǒng)需要將所有溢寫文件中的數(shù)據(jù)進(jìn)行歸并（從磁盤到磁盤以分區(qū)排序來歸并數(shù)據(jù)），以生成一個大的溢寫文件（數(shù)據(jù)已分區(qū)且有序）。歸并操作就是將相同key的value歸并成一個集合，形成新的鍵值對。例如，具有相同鍵的鍵值對<'a',1><'a',2><'a',5>，經(jīng)過歸并操作之后，得到的鍵值對為<'a',{1,2,5}>。文件歸并操作完成后生成最終的Map任務(wù)輸出文件，文件保存在Map任務(wù)所在節(jié)點的本地磁盤上，Map任務(wù)執(zhí)行結(jié)束。MapReduce的工作流程（1）在一個MapReduce作業(yè)中，通常會啟動多個Map任務(wù)，并且由于每個Map任務(wù)處理的數(shù)據(jù)量不同，任務(wù)結(jié)束時間也不同。不過，一旦有Map任務(wù)結(jié)束，與其相關(guān)的Reduce任務(wù)就會去復(fù)制輸出文件。系統(tǒng)會根據(jù)Reduce任務(wù)數(shù)來啟動相同數(shù)量的復(fù)制線程（Fetcher），這些復(fù)制線程能夠并行復(fù)制Map任務(wù)的輸出文件。（2）Reduce任務(wù)將復(fù)制獲得的文件存放在自身所在節(jié)點的緩存中，當(dāng)緩存中的數(shù)據(jù)達(dá)到閾值，即需要溢寫到磁盤時，Reduce任務(wù)會將復(fù)制數(shù)據(jù)進(jìn)行歸并排序（MergeSort），生成溢寫文件。如果生成了多個溢寫文件，則需要多次執(zhí)行歸并操作，再將數(shù)據(jù)輸入到reduce()函數(shù)。2．Reduce端的Shuffle過程MapReduce的工作流程6.2.4Reduce過程Reduce任務(wù)的輸出結(jié)果可以經(jīng)過輸出格式化后再輸出到文件系統(tǒng)中，并且每個作業(yè)輸出結(jié)果文件默認(rèn)以“part-r-00000”開始，并用后5位數(shù)遞增的方式命名。Reduce任務(wù)接收歸并排序后的數(shù)據(jù)流，并對已有序的相同key的鍵值對調(diào)用一次reduce()函數(shù)。其中，Reduce任務(wù)的輸入是<key2,list(value2)>形式的中間結(jié)果，輸出的是<key3,value3>形式的計算結(jié)果。03案例解析——單詞統(tǒng)計編程案例解析——單詞統(tǒng)計編程基于MapReduce的編程之所以難度不大，是因為MapReduce已經(jīng)為開發(fā)人員解決了諸如分布式存儲、負(fù)載均衡、容錯管理、網(wǎng)絡(luò)通信、工作調(diào)度等難題。開發(fā)人員只需分別實現(xiàn)Map任務(wù)和Reduce任務(wù)中的map()函數(shù)和reduce()函數(shù)即可，而這兩個函數(shù)來自MapReduceAPI提供的Mapper類和Reducer類。MapReduce主要處理HDFS中的文件，而HDFS中的文件格式多種多樣，且文件的大小也不固定，大的文件可達(dá)幾百GB，而小的文件卻只有幾個KB。因此，處理這樣的輸入數(shù)據(jù)時需要對其進(jìn)行格式化。此外，為了方便使用輸出結(jié)果，同樣也需要對輸出數(shù)據(jù)進(jìn)行格式化。除了輸入、輸出數(shù)據(jù)的格式化，還可以根據(jù)實際問題優(yōu)化數(shù)據(jù)處理過程，如數(shù)據(jù)分區(qū)、數(shù)據(jù)壓縮、數(shù)據(jù)合并（Map端的合并操作）及數(shù)據(jù)過濾等。案例解析——單詞統(tǒng)計編程6.3.1案例描述假設(shè)在hdfs://hadoop0:9000/jqe/wc/data目錄中存在兩個文本文件（文件中僅有字符串和空格），分別是words1.txt和words2.txt，現(xiàn)需計算出這兩個文件的單詞頻數(shù)。要求輸出結(jié)果按單詞的字母順序進(jìn)行排序，并輸出到兩個文件中（以字母“g”進(jìn)行拆分）。另外，還需要對輸出數(shù)據(jù)進(jìn)行格式化，要求單詞與詞頻之間以“~~~~~~”隔開。其中，words1.txt的內(nèi)容如下：thisisagooddogthatisagoodcatwords2.txt的內(nèi)容如下：thisisabaddogthatisabadcat案例解析——單詞統(tǒng)計編程輸出結(jié)果應(yīng)分為兩個文件（系統(tǒng)自動命名），分別是part-r-00000和part-r-00001。其中，part-r-00000的內(nèi)容如下：a~~~~~~4bad~~~~~~2cat~~~~~~2dog~~~~~~2part-r-00001的內(nèi)容如下：good~~~~~~2is~~~~~~4that~~~~~~2this~~~~~~2案例解析——單詞統(tǒng)計編程6.3.2設(shè)計思路在單詞統(tǒng)計任務(wù)中，可以將大的數(shù)據(jù)集切分成小的數(shù)據(jù)集，且各數(shù)據(jù)集之間相互獨立，方便并行處理。此外，各個單詞之間的頻數(shù)不具有相關(guān)性，可以將不同的單詞分發(fā)到不同的機(jī)器上處理。由此可以看出，單詞統(tǒng)計任務(wù)的解決思路完全貼合MapReduce的編程思想。根據(jù)分析，單詞統(tǒng)計程序可分為7個模塊，分別是輸入、輸入分片及其格式化、設(shè)計map()函數(shù)、分區(qū)、設(shè)計reduce()函數(shù)、輸出格式化、輸出。案例解析——單詞統(tǒng)計編程指定輸入文件位置。將兩個文件切分成兩個輸入分片，然后對輸入分片進(jìn)行格式化，如文件按行分解，從而形成以key為偏移量、value為行內(nèi)容的鍵值對。設(shè)計map()函數(shù)，將鍵值對的value按空格分解成一個個單詞（word），生成<word,1>形式的鍵值對。設(shè)計分區(qū)函數(shù)，將key的首字母小于“g”的鍵值對劃分為一個分區(qū)，其余的劃分為另外一個分區(qū)。設(shè)計reduce()函數(shù)，將輸入的<word,list(value)>形式（如<"is",{1,1}>）的數(shù)據(jù)進(jìn)行匯總，生成以key為單詞、value為單詞頻數(shù)的鍵值對。系統(tǒng)默認(rèn)的輸出格式為“單詞+空格+單詞頻數(shù)”的形式，若要輸出特定樣式的數(shù)據(jù)，需要對輸出結(jié)果進(jìn)行格式化，使得單詞與單詞頻數(shù)之間以“~~~~~~”間隔。指定單詞統(tǒng)計程序的輸出結(jié)果文件位置。案例解析——單詞統(tǒng)計編程6.3.3執(zhí)行過程在給出的案例中，輸入文件為兩個很小的文本文件，單個文件的數(shù)據(jù)沒有達(dá)到需要切分的程度，所以可將每個文件作為獨立的分片。此外，還需要對輸入分片進(jìn)行格式化操作，形成<key1,value1>形式的數(shù)據(jù)流。1．輸入分片及其格式化單詞統(tǒng)計的輸入分片及其格式化key1為偏移量，從0開始，每讀取一個字符（包括空格）就增加1；value1為每行文本內(nèi)容（字符串形式）。案例解析——單詞統(tǒng)計編程map()函數(shù)將接收到的<key1,value1>形式的輸入數(shù)據(jù)流，按空格進(jìn)行拆分，輸出結(jié)果為<key2,value2>形式的數(shù)據(jù)。2．Map過程key2為字符串形式的單詞；value2的值為1，表示單詞數(shù)為1。單詞統(tǒng)計的Map過程案例解析——單詞統(tǒng)計編程由于Reduce要求輸入數(shù)據(jù)有序，所以map()函數(shù)的計算結(jié)果需要經(jīng)過處理（如分區(qū)、排序、歸并），才可以作為reduce()函數(shù)的輸入。于是，將多個Map任務(wù)的<key2,value2>形式的輸出，處理成<key2,list(value2)>形式的中間結(jié)果3．Shuffle過程單詞統(tǒng)計的Shuffle過程案例解析——單詞統(tǒng)計編程reduce()函數(shù)接收<key2,list(value2)>形式的數(shù)據(jù)流，對相同單詞的值集合進(jìn)行計算，匯總出單詞出現(xiàn)的總次數(shù)。4．Reduce過程單詞統(tǒng)計的Reduce過程案例解析——單詞統(tǒng)計編程用戶可以根據(jù)需求，更改單詞統(tǒng)計結(jié)果的輸出格式。5．輸出單詞統(tǒng)計的輸出結(jié)果格式化案例解析——單詞統(tǒng)計編程6.3.4編程實現(xiàn)在Eclipse開發(fā)工具中新建一個Map/Reduce項目“WordCount”（即單詞統(tǒng)計項目），然后根據(jù)單詞統(tǒng)計的設(shè)計思路進(jìn)行MapReduce編程。使用MapReduceAPI中的org.apache.hadoop.mapreduce.lib.input.TextInputFormat進(jìn)行輸入格式化；自定義WordCountMapper類實現(xiàn)單詞統(tǒng)計的map()函數(shù)；自定義WordCountPartitioner類實現(xiàn)對單詞統(tǒng)計結(jié)果的分區(qū)存儲；自定義WordCountReducer類實現(xiàn)單詞統(tǒng)計的reduce()函數(shù)；自定義WCFileOutputFormat類實現(xiàn)對數(shù)據(jù)結(jié)果的格式化。編寫WordCountDriver類（驅(qū)動類）去調(diào)用其他類定義的方法。案例解析——單詞統(tǒng)計編程自定義的WordCountMapper類繼承了org.apache.hadoop.mapreduce.Mapper類。WordCountMapper類是map()函數(shù)的執(zhí)行者，用戶需要根據(jù)數(shù)據(jù)處理需求重寫map()方法。1．自定義WordCountMapper類創(chuàng)建WordCountMapper類定義了值為1的“one”常量，表示每個單詞都是統(tǒng)計一次，然后又定義了一個“words”字符串?dāng)?shù)組，用于存放每行分解的單詞，最后將“one”和“words”作為參數(shù)傳給write()方法，相當(dāng)于map()函數(shù)的一次輸出。WordCountMapper類的代碼參照課本。案例解析——單詞統(tǒng)計編程自定義的WordCountPartitioner類繼承了org.apache.hadoop.mapreduce.Partitioner類，然后重寫getPartition()方法。該方法中使用了正則表達(dá)式，并按key對map()函數(shù)輸出的鍵值對進(jìn)行劃分。例如，將key的字母小于“g”的鍵值對發(fā)送給一個Reduce任務(wù)，其他的鍵值對發(fā)送到另一個Reduce任務(wù)。WordCountPartitioner類的代碼參照課本。2．自定義WordCountPartitioner類自定義的WordCountReducer類繼承了org.apache.hadoop.mapreduce.Reducer類，并重寫了reduce()方法。相同單詞的鍵值對會在同一個Reduce任務(wù)中處理，該方法中定義了一個“count”整型變量，循環(huán)遍歷一個鍵值對的valuelist，用于保存value累加的結(jié)果。WordCountReducer類的代碼參照課本。3．自定義WordCountReducer類案例解析——單詞統(tǒng)計編程自定義的WCFileOutputFormat類繼承了org.apache.hadoop.mapreduce.lib.output.FileOutputFormat類，并重寫了getRecordWriter()方法。該方法中定義了一個文件輸出流對象，用于將結(jié)果輸出到HDFS文件中。WCFileOutputFormat類的代碼參照課本。4．自定義WCFileOutputFormat類自定義的WordCountDriver類為Job指定了輸入文件位置、輸入格式化類、自定義Partitioner類、自定義Mapper類、自定義Reducer類、自定義OutputFormat類及輸出文件位置，其代碼參照課本。5．自定義WordCountDriver類案例解析——單詞統(tǒng)計編程6.3.5運行程序運行WordCount程序有兩種方法：利用EclipseHadoop插件在本地提交任務(wù)到集群上運行；將編寫完成的程序打包成jar包并上傳到集群上運行。1．利用EclipseHadoop插件1編寫一個perties文件（用于記錄MapReduce的運行日志），并將其放入WordCount項目的“src”文件夾，然后在Eclipse中手動刷新一下。案例解析——單詞統(tǒng)計編程2右擊WordCountDriver類，從彈出的快捷菜單中選擇“RunAs”→“RunonHadoop”選項，以該方式編譯運行程序。程序運行完成后，會在“hdfs://hadoop0:9000/jqe/wc/out/”路徑下生成相關(guān)文件：_SUCCESS、part-r-00000和part-r-00001，單詞統(tǒng)計結(jié)果如下圖所示。本地提交任務(wù)到集群上運行的結(jié)果案例解析——單詞統(tǒng)計編程2．將項目打包成jar包1將WordCount項目打包成一個jar包：在“ProjectExplorer”窗口中右擊“WordCount”項目，從彈出的快捷菜單中選擇“Export...”選項，打開“Export”對話框，選擇“Java”→“JARfile”選項，然后單擊“Next”按鈕，彈出“JARExport”對話框，指定jar包的保存位置和名稱（如E:\WCMR.jar），然后單擊“Finish”按鈕。2通過WinSCP軟件將jar包“WCMR.jar”上傳到集群節(jié)點（如hadoop0）的本地目錄（如“/opt”）。案例解析——單詞統(tǒng)計編程3在hadoop0節(jié)點中執(zhí)行以下命令，運行WordCount程序：#hadoopjar/opt/WCMR.jardriver.WordCountDriverWordCount程序在集群上的運行過程如下圖所示。單詞統(tǒng)計結(jié)果與第1種方法相同。WordCount程序在集群上的運行過程案例解析——單詞統(tǒng)計編程6.3.6數(shù)據(jù)優(yōu)化數(shù)據(jù)壓縮對于海量數(shù)據(jù)處理具有重要的意義。在MapReduce作業(yè)中，存在數(shù)據(jù)壓縮的情況有3種：1．?dāng)?shù)據(jù)壓縮①輸入數(shù)據(jù)可以是壓縮文件；②Map輸出的結(jié)果可以是壓縮數(shù)據(jù)；③Reduce輸出的最終結(jié)果可以是壓縮文件。MapReduce程序的輸入、輸出是壓縮文件易于理解。另外，若用戶選擇在Map階段對數(shù)據(jù)進(jìn)行壓縮，就能夠減少在Reduce階段復(fù)制Map數(shù)據(jù)結(jié)果的數(shù)據(jù)量。但值得注意的是，數(shù)據(jù)壓縮不適合小文件多的MapReduce任務(wù)。案例解析——單詞統(tǒng)計編程設(shè)置Map輸出為壓縮數(shù)據(jù)有兩種方法。在配置文件mapred-site.xml中設(shè)置壓縮參數(shù)，具體可通過配置press屬性（是否開啟壓縮）和press.codec屬性（指定壓縮格式）來進(jìn)行。在Job中聲明，如在WordCountDriver類中添加如下代碼：conf.setBoolean("press",true);conf.setClass("press.codec",BZip2Codec.class,CompressionCodec.class);案例解析——單詞統(tǒng)計編程2．Map端的合并操作Map端的合并操作沒有默認(rèn)的實現(xiàn)，需要在Job中顯式設(shè)置才能起作用。如果設(shè)置了Map端的合并操作，那么每個運行的Map任務(wù)中都會執(zhí)行合并操作，且僅僅只處理自身節(jié)點生成的數(shù)據(jù)。Map端的合并操作相當(dāng)于對輸入Reduce任務(wù)的數(shù)據(jù)進(jìn)行預(yù)處理，即將相同key的鍵值對合并。如此一來，Map任務(wù)的輸出數(shù)據(jù)更緊湊且數(shù)據(jù)量變小，從而減少了Reduce任務(wù)復(fù)制Map任務(wù)計算結(jié)果的網(wǎng)絡(luò)帶寬和Reduce上的負(fù)載。若希望在單詞統(tǒng)計程序中設(shè)置Map端的合并操作，可在WordCountDriver類中增加以下代碼：job.setCombinerClass(WordCountReducer.class);案例解析——單詞統(tǒng)計編程下圖顯示了在MapReduce的Shuffle過程中設(shè)置了Map端的合并操作后的數(shù)據(jù)流的形式，顯然，它與未設(shè)置合并操作的數(shù)據(jù)流形式有所不同。

配置Map端合并操作的Shuffle過程根據(jù)實際數(shù)據(jù)處理任務(wù)的不同，數(shù)據(jù)過濾操作使用戶可以只針對有價值的數(shù)據(jù)進(jìn)行計算，從而大大減少數(shù)據(jù)的計算量。尤其是針對海量數(shù)據(jù)進(jìn)行處理時，此項功能非常有用。3．?dāng)?shù)據(jù)過濾案例解析——單詞統(tǒng)計編程在統(tǒng)計單詞的MapReduce程序中，可以通過對Map任務(wù)的輸出結(jié)果進(jìn)行過濾來減少進(jìn)入Reduce任務(wù)的計算量，如不統(tǒng)計“am”“is”“are”這樣的詞。為此，可更改WordCountMapper類的map()方法，代碼如下：publicvoidmap(LongWritablek1,Textv1,Mapper<LongWritable,Text,Text,IntWritable>.Contextcontext)throwsIOException,InterruptedException{ Stringline=v1.toString(); //獲得行

String[]words=line.split(""); //按空格切分行，獲得單詞數(shù)組

List<String>passwords=newArrayList<String>(); passwords.add("am"); passwords.add("is"); passwords.add("are"); for(Stringw:words){ //輸出鍵值對<單詞,1> word.set(w); if(!passwords.contains(word)){ context.write(word,one); } }}04案例解析——二次排序編程案例解析——二次排序編程MapReduce在傳遞<key,value>鍵值對時，默認(rèn)按照key進(jìn)行排序，但有時候除了key以外，還需要根據(jù)value或value中的某一個字段進(jìn)行排序，基于這種需求進(jìn)行的自定義排序稱為“二次排序”。案例解析——二次排序編程6.4.1案例描述假設(shè)文本文件user.txt中存放著某公司的職工信息，每行均包含兩個字符串，并以空格分隔。其中，第1個字符串表示部門ID，第2個字符串表示職工姓名?，F(xiàn)要求先按部門升序排列，然后相同部門的職工姓名以字典序排列，并且部門之間以“------------------”分隔開。6.4.2設(shè)計思路Map任務(wù)的輸出按key進(jìn)行排序，如果將本例中的第1個屬性作為Map任務(wù)輸出鍵值對的key，第2個屬性作為value，那么二次排序任務(wù)還要對value排序，但MapReduce沒有提供該方法。因此，可以將key和value結(jié)合起來，形成newkey組合鍵，設(shè)置value為NullWritable類型的占位符，則Map任務(wù)的輸出結(jié)果為<newkey,null>，即<<key,value>,null>。案例解析——二次排序編程二次排序的過程分為輸入、輸入分片及其格式化、Map過程、排序過程、分組過程、Reduce過程和輸出過程。（1）二次排序程序的輸入文件只有一個，因此可將user.txt文件作為一個輸入分片，然后進(jìn)行格式化。二次排序的輸入分片及其格式化案例解析——二次排序編程（2）map()函數(shù)將接收到的輸入數(shù)據(jù)流按行內(nèi)容以空格進(jìn)行拆分，輸出結(jié)果為<<key,value>,null>形式的數(shù)據(jù)。其中，key為文本中第1個屬性，value為文本中第2個屬性。二次排序的Map過程（3）將map()函數(shù)的輸出鍵值對按key進(jìn)行排序，當(dāng)key相同時，按value進(jìn)行排序。二次排序的map()函數(shù)結(jié)果排序案例解析——二次排序編程（4）在Reduce端設(shè)置分組方法，按key進(jìn)行分組。二次排序的分組操作（5）red