單元五大數(shù)據(jù)的離線處理MapReduce掌握Map和Reduce階段的作用和創(chuàng)建掌握MapReduceJava開發(fā)項(xiàng)目的搭建熟悉HadoopWritable中的各種數(shù)據(jù)類型熟悉MapReduce開發(fā)環(huán)境安裝配置熟悉Shuffle、Partition、Combiner等組件的作用理解實(shí)時(shí)處理與離線處理5.1離線處理技術(shù)MapReduce

MapReduce是Hadoop平臺(tái)自帶的一個(gè)運(yùn)算框架，安裝Hadoop，也就安裝了MapReduce（以下簡稱MR）。MR采用的是“分而治之，然后匯總”的算法思想，首先將數(shù)據(jù)拆分成若干個(gè)小的數(shù)據(jù)，運(yùn)算這些小數(shù)據(jù)的任務(wù)過程在MR中被稱為map階段；然后匯總結(jié)果形成一個(gè)最終的運(yùn)算結(jié)果，這一過程在MR中被稱為reduce階段。

MR在進(jìn)行數(shù)據(jù)處理時(shí)，會(huì)將運(yùn)算任務(wù)視為一個(gè)job（作業(yè)），job的生命周期從任務(wù)提交開始，到任務(wù)完成，退出MR時(shí)結(jié)束。因此，如果有新的數(shù)據(jù)需要處理時(shí)，必須重新提交。這種任務(wù)模型只適合于對時(shí)效要求不高的離線型運(yùn)算任務(wù)。

每個(gè)階段都采用鍵值對（key/value）方式作為數(shù)據(jù)的輸入（input）和輸出（output），其基本的數(shù)據(jù)模型如下。

(input)<k1,v1>→map→<k2,v2>→reduce→<k3,v3>(output)開發(fā)人員的任務(wù)，就是利用MapReduce運(yùn)算框架，在兩個(gè)階段所對應(yīng)的map()和reduce()方法中，按業(yè)務(wù)需求進(jìn)行數(shù)據(jù)的處理。5.2配置Hadoop開發(fā)環(huán)境 5.2.1下載Hadoop

訪問Hadoop下載地址“/releases.html”（截止到2021年8月，Hadoop的最新穩(wěn)定版是3.3.1），下載Hadoop的安裝包。將此安裝包解壓到Windows系統(tǒng)的某個(gè)不包含中文的空目錄中。5.2.2Hadoop插件配置 將hadoop-eclipse-plugin-2.7.3.jar文件復(fù)制到eclipse安裝目錄下的“dropins”目錄。1.eclipse開發(fā)工具的配置

啟動(dòng)eclipse工具，單擊“Window”→“Preferences”菜單項(xiàng)。在“Preferens”窗口左側(cè)，單擊“HadoopMap/Reduce”菜單，，將解壓后的hadoop目錄引入。單擊“Window菜單”→“ShowView”菜單列表→“Other”菜單項(xiàng)，選擇“Map/ReduceLocations”圖標(biāo)。在eclipse的下方輸出窗口區(qū)，找到“Map/ReduceLocations”工作窗口，單擊該窗口右上角的“NewHadoopLocation”圖標(biāo)。在彈出的窗口中依次填寫Windows系統(tǒng)的IP地址、Hadoop集群中處于active狀態(tài)的節(jié)點(diǎn)主機(jī)名或IP地址、NameNode的遠(yuǎn)程訪問端口號，以及Hadoop集群的用戶賬號，在進(jìn)行操作前，應(yīng)啟動(dòng)好HDFS+Yarn集群。在eclipse的“ProjectExplorer”窗口，單擊“DFSLocations”圖標(biāo)→選擇其下出現(xiàn)的子圖標(biāo)→單擊鼠標(biāo)右鍵，在快捷菜單中，單擊“Reconnect”菜單項(xiàng)，去連接Hadoop集群，單擊“Refresh”菜單項(xiàng)可以刷新連接信息。連接成功后，Hadoop集群上的文件目錄結(jié)構(gòu)會(huì)顯示在窗口中將資源包中以下兩個(gè)文件放置到Windows系統(tǒng)的Hadoop解壓目錄的bin子目錄中。2.Windows系統(tǒng)環(huán)境變量的配置新建一個(gè)環(huán)境變量，在“編輯系統(tǒng)變量”窗口，在“變量名”輸入框中填寫“HADOOP_HOME”，在“變量值”輸入框中填寫Hadoop在Windows系統(tǒng)中的解壓路徑。編輯“Path”環(huán)境變量，在“編輯系統(tǒng)變量”窗口的“變量值”輸入框的最末尾處，輸入;%HADOOP_HOME%\bin;注意輸入的代碼內(nèi)容前后有一對英文的分號包圍，最后單擊“確定”按鈕，退出環(huán)境變量設(shè)置。5.3創(chuàng)建MR項(xiàng)目 打開eclipse工具，單擊“File”菜單→指向“New”菜單列表→單擊“Project”菜單項(xiàng)。在彈出的“NewProject”窗口中，選擇“Map/ReduceProject”圖標(biāo)后。將在5.2.1節(jié)解壓后的Hadoop包下的“etc/hadoop”目錄中的perties文件，復(fù)制到項(xiàng)目的“src”目錄中。MR項(xiàng)目中以鍵值對（key/value）來組織和傳遞數(shù)據(jù)，其鍵值對類型是HadoopWritable接口中的數(shù)據(jù)類型如下表，表中同時(shí)列舉了Hadoop數(shù)據(jù)類型對應(yīng)的Java數(shù)據(jù)類型。序號Hadoop數(shù)據(jù)類型Java數(shù)據(jù)類型1ArrayWritableArray2BooleanWritableBoolean3ByteWritableByte4DoubleWritableDouble5FloatWritableFloat6IntWritableint/char7LongWritableLong8ShortWritableShort9TextString10NullWritableNull5.4采用MR框架實(shí)現(xiàn)一個(gè)簡易的詞頻統(tǒng)計(jì)程序

本項(xiàng)目的算法模型見下圖，在MR的map階段，默認(rèn)存在有shuffle（洗牌）過程。//創(chuàng)建繼承于Mapper的子類，對指定的HDFS文件中的數(shù)據(jù)進(jìn)行拆分publicstaticclassWordCountMapextendsMapper<LongWritable,Text,Text,IntWritable>{//實(shí)現(xiàn)基類的map方法protectedvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{String[]strs=value.toString().split("");//進(jìn)行句子內(nèi)容的拆分for(Stringstr:strs){//將拆分后的每個(gè)單詞，以及數(shù)量傳遞給下游的Reduce，每個(gè)拆分出的單詞，都給予數(shù)量值1context.write(newText(str),newIntWritable(1));}}}//創(chuàng)建繼承于Reducer的子類，對接收到的單詞進(jìn)行分類統(tǒng)計(jì)publicstaticclassWordCountReduceextendsReducer<Text,IntWritable,Text,IntWritable>{protectedvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{intsum=0;for(IntWritableval:values){//遍歷集合中的每個(gè)成員sum+=val.get();//從成員值中獲取單詞的數(shù)量，并進(jìn)行累加求和}context.write(key,newIntWritable(sum));//將統(tǒng)計(jì)結(jié)果寫入HDFS}}分別創(chuàng)建Map和Reduce環(huán)節(jié)對應(yīng)的實(shí)現(xiàn)類publicstaticvoidmain(String[]args)throwsException{//指定MR程序的數(shù)據(jù)來源與運(yùn)算結(jié)果的存放位置,按要求填寫主機(jī)名或IP地址StringinputPath="hdfs://Hadoop集群中處于active狀態(tài)的節(jié)點(diǎn)主機(jī)名或IP地址:9820/input";StringoutputPath="hdfs://Hadoop集群中處于active狀態(tài)的節(jié)點(diǎn)主機(jī)名或IP地址:9820/output";Configurationconf=newConfiguration();//創(chuàng)建Hadoop配置對象Jobjob=Job.getInstance(conf);//創(chuàng)建MR的Job任務(wù)工作對象job.setJarByClass(HelloWorld.class);job.setOutputKeyClass(Text.class);//分別設(shè)置MR運(yùn)算結(jié)果的鍵與值的數(shù)據(jù)類型job.setOutputValueClass(IntWritable.class);job.setMapperClass(WordCountMap.class);//為Job配置Map和Reducejob.setReducerClass(WordCountReduce.class);//為Job配置數(shù)據(jù)來源與運(yùn)算結(jié)果的存放位置FileInputFormat.addInputPath(job,newPath(inputPath));FileOutputFormat.setOutputPath(job,newPath(outputPath));job.waitForCompletion(true);//阻塞主進(jìn)程，直至MR的Job任務(wù)運(yùn)行結(jié)束}創(chuàng)建任務(wù)Job，組裝Map和Reduce實(shí)現(xiàn)類，并配置輸入/輸出地址等。5.5采用MR框架進(jìn)行中文詞頻統(tǒng)計(jì)

本節(jié)項(xiàng)目的特點(diǎn)：(1)在Map階段，對三個(gè)不同時(shí)期的中文小說，進(jìn)行分片；(2)采用第三方庫jieba對分片的句子進(jìn)行符合語意的詞匯拆分；(3)對拆分后的詞匯進(jìn)行過濾，剔除非中文字符及符號；(4)對過濾后的詞匯，按文章進(jìn)行Partition（分區(qū)），匯聚到各自對應(yīng)的分區(qū)；(5)在Combiner階段，對各分區(qū)中的key進(jìn)行清洗，剔除key中包含的小說名稱，并進(jìn)行詞頻統(tǒng)計(jì)；(6)在Reduce階段，匯總詞頻統(tǒng)計(jì)，形成三個(gè)小說各自的三個(gè)詞頻統(tǒng)計(jì)結(jié)果文件；(7)采用Sqoop將三個(gè)小說的詞頻統(tǒng)計(jì)結(jié)果導(dǎo)出到MySQL中；(8)采用Matplotlib子圖的方式對三個(gè)小說中排名前10的詞頻進(jìn)行數(shù)據(jù)對比可視化。Partition（分區(qū)）是MR框架根據(jù)給定的條件，將Map階段產(chǎn)生的鍵值對（key/value）數(shù)據(jù)分配到指定的Reduce中進(jìn)行處理的過程。因此，Partition（分區(qū)）數(shù)量要求與Reduce的數(shù)量保持一致，同時(shí)，在代碼中要求通過以下代碼，明確設(shè)置Reduce的數(shù)量，只有當(dāng)設(shè)置的Reduce數(shù)量大于1時(shí)，Partition功能才能生效。setNumReduceTasks(inttasks)MR框架的Combiner階段，實(shí)質(zhì)就是一個(gè)預(yù)Reduce階段，其作用就是對Map階段產(chǎn)生的的鍵值對（key/value）數(shù)據(jù)預(yù)先做一次匯總，減少Reduce階段的匯總次數(shù)，從而達(dá)到提高Reduce階段的執(zhí)行效率，Combiner階段的匯總結(jié)果將作為Reduce階段的輸入。例如，圖中的Partition（分區(qū)）標(biāo)準(zhǔn)，是將以h字母開頭的單詞歸為一組，因頁面篇幅有限，圖中省略了Input環(huán)節(jié)，從圖中可以看出，Combiner階段并非總是必須的，它的使用，必須根據(jù)業(yè)務(wù)需求來進(jìn)行設(shè)置。5.5.1開發(fā)前的準(zhǔn)備工作由于Hadoop的Text類使用的是UTF-8編碼格式，因此在項(xiàng)目開發(fā)前，需要對小說的中文字符編碼從默認(rèn)的ANSI編碼，轉(zhuǎn)換成UTF-8編碼。在Windows系統(tǒng)中，可以用記事本軟件調(diào)整編碼格式。上傳小說素材文件到HDFS目錄中5.5.2項(xiàng)目實(shí)現(xiàn)

配置好后的MR項(xiàng)目整體結(jié)構(gòu)項(xiàng)目中添加了以下內(nèi)容：(1)包名：com.my.novels.wordscount；(2)App.java文件：項(xiàng)目中唯一的java文件；(3)perties文件：參見教材5.4.2節(jié)中的介紹；(4)lib新目錄，及jieba庫的jar文件。......publicstaticclassNovelsMapextendsMapper<LongWritable,Text,Text,IntWritable>{......@Overrideprotectedvoidsetup(Contextcontext)throwsIOException,InterruptedException{this.segmenter=newJiebaSegmenter();//設(shè)置保留中文字符作用的正則表達(dá)式this.pattern=Ppile("[\u4E00-\u9FA5]+");}......在App.java文件中setup（）方法在Mapper類中主要起初始化作用，可以用來做業(yè)務(wù)需求上的一些初始化工作publicstaticclassNovelsPartitionerextendsPartitioner<Text,IntWritable>{@OverridepublicintgetPartition(Textkey,IntWritablevalue,intnumPartitions){StringfileName=key.toString().split(":")[0];if(fileName.equals("《平凡的世界》作者：路遙.txt")){return0;}elseif(fileName.equals("朝花夕拾.txt")){return1;}else{return2;}}}進(jìn)行分區(qū)操作，將不同小說的詞匯分配到對應(yīng)的Reduce中。在App.java文件中publicstaticclassNovelsCombinerextendsReducer<Text,IntWritable,Text,IntWritable>{@Overrideprotectedvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{

......if(fileName.equals("《平凡的世界》作者：路遙.txt")){//將詞匯+空格+代表小說的編號作為key，方便后面的Sqoop導(dǎo)出到MySQL中context.write(newText(word+""+1),newIntWritable(sum));}elseif(fileName.equals("朝花夕拾.txt")){context.write(newText(word+""+2),newIntWritable(sum));}else{context.write(newText(word+""+3),newIntWritable(sum));}}Combiner類的創(chuàng)建，對即將進(jìn)入最后Reduce階段的數(shù)據(jù)，提前進(jìn)行分類加工，從key中剔除小說名稱，添加小說ID號，以符合后續(xù)的MySQL表結(jié)構(gòu)的設(shè)計(jì)。在App.java文件中publicstaticvoidmain(String[]args)throwsException{

......//屏蔽MR默認(rèn)的空格間隔符，設(shè)置以英文的逗號作為key和value之間的間隔符，便于sqoop識(shí)別conf.set("mapred.textoutputformat.ignoreseparator","true");conf.set("mapred.textoutputformat.separator",",");

Jobjob=Job.getInstance(conf);job.setJarByClass(App.class);

job.setMapperClass(NovelsMap.class);job.setPartitionerClass(NovelsPartitioner.class);//設(shè)置分區(qū)job.setCombinerClass(NovelsCombiner.class);//設(shè)置Combinerjob.setReducerClass(NovelsReduce.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);job.setNumReduceT