大數(shù)據(jù)之MapReduce應(yīng)用

MapReduce應(yīng)用云計算與大數(shù)據(jù)概論目標(biāo)了解MapReduce計算模型的基本原理和工作流程了解Hadoop中實現(xiàn)并行計算的相關(guān)機制掌握MapReduce的任務(wù)調(diào)度過程掌握MapReduce的詳細(xì)執(zhí)行過程了解MapReduce新框架YARN的原理及運作機制MapReduce模型概述MapReduce是Google公司的核心計算模型，它將復(fù)雜的、運行于大規(guī)模集群上的并行計算過程高度地抽象到了兩個函數(shù)——Map和Reduce——之中。一個MapReduce作業(yè)（Job）通常會把輸入的數(shù)據(jù)集切分為若干獨立的數(shù)據(jù)塊，由Map任務(wù)（Task）以完全并行的方式進(jìn)行處理。MapReduce框架會對Map的輸出先進(jìn)行排序，然后把結(jié)果輸入給Reduce任務(wù)，通常作業(yè)的輸入和輸出都會被存儲在文件系統(tǒng)中，而整個框架負(fù)責(zé)任務(wù)的調(diào)度和監(jiān)控，并重新執(zhí)行已經(jīng)失敗的任務(wù)。MapReduce計算模型的核心是Map和Reduce兩個函數(shù)，這兩個函數(shù)由用戶負(fù)責(zé)實現(xiàn)，作用是按照一定的映射規(guī)則，將輸入的鍵值對轉(zhuǎn)換成另一個或一批鍵值對并輸出?；贛apReduce計算模型編寫分布式并行程序非常簡單，程序員只需負(fù)責(zé)Map和Reduce函數(shù)的主要編碼工作，而并行編程中的其它種種復(fù)雜問題，諸如分布式存儲、工作調(diào)度、負(fù)載均衡、容錯處理、網(wǎng)絡(luò)通信等，均可由MapReduce框架（如Hadoop）代為處理，程序員完全不用操心。工作組件MapReduce的運行機制中主要包含以下幾個獨立的大類組件：Client：主要作用是提交MapReduce作業(yè)并為用戶顯示處理結(jié)果。ResourceManager：主要進(jìn)行MapReduce作業(yè)執(zhí)行的協(xié)調(diào)工作，是MapReduce運行機制中的主控節(jié)點。NameNode：文件管理系統(tǒng)中的中心服務(wù)器，負(fù)責(zé)管理文件系統(tǒng)的命名空間（元數(shù)據(jù)），維護(hù)整個文件系統(tǒng)的文件目錄樹以及這些文件的索引目錄，并記錄文件和目錄的擁有者和權(quán)限、文件包含的數(shù)據(jù)塊、塊的個數(shù)以及塊的副本數(shù)，以及決定數(shù)據(jù)塊（Block）到具體DataNode節(jié)點的映射等。ApplicationMaster：管理在YARN內(nèi)運行的應(yīng)用程序?qū)嵗瑓f(xié)調(diào)來自ResourceManager的資源。NodeManager：每一臺機器框架的代理，是執(zhí)行應(yīng)用程序的容器，監(jiān)控應(yīng)用程序的資源使用情況（如CPU，內(nèi)存，硬盤，網(wǎng)絡(luò)等）并向調(diào)度器匯報。DataNode：負(fù)責(zé)處理文件系統(tǒng)的讀寫請求，在NameNode的指揮下進(jìn)行數(shù)據(jù)塊的創(chuàng)建、刪除和復(fù)制。Container：YARN中資源的抽象，封裝了某個節(jié)點上一定量的資源（CPU和內(nèi)存兩類資源）MapReduce工作流程MapReduce處理大數(shù)據(jù)集的計算過程就是將大數(shù)據(jù)集分解為成百上千的小數(shù)據(jù)集，每個（或若干個）數(shù)據(jù)集分別由集群中的一個節(jié)點（通常就是一臺普通的計算機）進(jìn)行處理并生成中間結(jié)果，然后這些中間結(jié)果又由大量的節(jié)點進(jìn)行合并，從而形成最終結(jié)果，如圖MapReduce各個執(zhí)行階段Hadoop中一個簡單的MapReduce任務(wù)執(zhí)行流程如下：（1）JobTracker負(fù)責(zé)在分布式環(huán)境中實現(xiàn)客戶端任務(wù)的創(chuàng)建和提交；（2）InputFormat模塊負(fù)責(zé)進(jìn)行Map前的預(yù)處理。（3）將RecordReader處理后的結(jié)果作為Map的輸入數(shù)據(jù)，然后由Map執(zhí)行預(yù)先定義的Map邏輯，將處理后的鍵值對結(jié)果輸出到臨時中間文件。（4）在MapReduce流程中，為了讓Reduce能并行處理Map結(jié)果，必須對Map的輸出結(jié)果進(jìn)行一定的排序和分割，然后再交給對應(yīng)的Reduce，而這個將Map輸出作進(jìn)一步整理并交給Reduce的過程，就稱為Shuffle。（5）Reduce執(zhí)行具體的業(yè)務(wù)邏輯，即用戶編寫的處理數(shù)據(jù)并得到結(jié)果的業(yè)務(wù)，并且將處理結(jié)果輸出給OutputFormat。Shuffle過程詳解Shuffle過程是MapReduce工作流程的核心，Shuffle過程包含在Map和Reduce兩端中，在Map端，Shuffle過程對Map的結(jié)果進(jìn)行劃分（Partition）、排序（Sort）和溢寫（Spill），然后將屬于同一劃分的輸出結(jié)果合并在一起并寫到磁盤上。而在Reduce端，Shuffle又會將各個Map送來的屬于同一劃分的輸出結(jié)果進(jìn)行合并（Merge），然后對合并的結(jié)果進(jìn)行排序，最后交給Reduce處理。Map端的Shuffle過程Map端的Shuffle過程簡單地說，每個MapTask都有一個內(nèi)存緩沖區(qū)，存儲著Map的輸出結(jié)果，當(dāng)緩沖區(qū)快滿的時候，需要將緩沖區(qū)的數(shù)據(jù)以一個臨時文件的方式存放到磁盤，當(dāng)整個MapTask結(jié)束后，再對磁盤中這個MapTask產(chǎn)生的所有臨時文件做合并，生成最終的正式輸出文件，然后等待ReduceTask來取數(shù)據(jù)。Reduce端的Shuffle過程ReduceTask前面的工作就是不斷地拉取當(dāng)前Job里每個MapTask的最終結(jié)果，然后對從不同地方拉取過來的數(shù)據(jù)不斷地執(zhí)行Merge（整合）操作，最終形成一個文件作為ReduceTask的輸入文件，如圖并行計算的實現(xiàn)-1數(shù)據(jù)分布存儲Hadoop中的分布式文件系統(tǒng)HDFS由一個管理節(jié)點（NameNode）和N個數(shù)據(jù)節(jié)點（DataNode）組成，每個節(jié)點均是一臺普通的計算機。在使用方面與我們熟悉的單機文件系統(tǒng)非常類似，同樣可以新建目錄、創(chuàng)建、復(fù)制、刪除文件、以及查看文件內(nèi)容等。但其在底層實現(xiàn)方面是把文件切割成Block（數(shù)據(jù)塊），然后將這些Block分散存儲在不同的DataNode上，每個Block還可以復(fù)制數(shù)份存儲于不同的DataNode上，達(dá)到容錯容災(zāi)的目的。NameNode則是整個HDFS的核心，它通過維護(hù)一些數(shù)據(jù)結(jié)構(gòu)，記錄下每一個文件被切割成了多少個Block，這些Block可以從哪些DataNode中獲得以及各個DataNode的狀態(tài)等重要信息。并行計算的實現(xiàn)-2分布式并行計算Hadoop中有的JobTracker用于調(diào)度和管理其它的TaskTracker。JobTracker可以運行在集群中任一臺計算機上，但負(fù)責(zé)執(zhí)行任務(wù)的TaskTracker必須運行于DataNode上，即DataNode既是數(shù)據(jù)存儲節(jié)點，也是計算節(jié)點。MapReduce任務(wù)請求調(diào)度的過程包括：JobClient提交作業(yè)和JobTacker調(diào)度作業(yè)。并行計算的實現(xiàn)-3本地計算本地計算，即數(shù)據(jù)存儲在哪一臺計算機上，就由這臺計算機進(jìn)行這部分?jǐn)?shù)據(jù)的計算，這樣可以減少數(shù)據(jù)在網(wǎng)絡(luò)上的傳輸，降低對網(wǎng)絡(luò)帶寬的需求。在Hadoop這樣的基于集群的分布式并行系統(tǒng)中，計算節(jié)點可以很方便地擴(kuò)充，因為它所能夠提供的計算能力近乎是無限的，但是，由于數(shù)據(jù)需要在不同的計算機之間流動，故而對網(wǎng)絡(luò)帶寬的要求比較高，這種情況下，本地計算就是最有效的一種節(jié)約網(wǎng)絡(luò)帶寬的手段，業(yè)界將其形容為“移動計算比移動數(shù)據(jù)更經(jīng)濟(jì)”。并行計算的實現(xiàn)-4任務(wù)粒度把原始大數(shù)據(jù)集切分成小數(shù)據(jù)集時，需要控制好切分粒度，通常是讓小數(shù)據(jù)集小于或等于HDFS中一個數(shù)據(jù)塊的大?。J(rèn)是64M），這樣能夠保證一個小數(shù)據(jù)集位于一臺計算機上，便于進(jìn)行本地計算。如果有M個小數(shù)據(jù)集待處理，就啟動M個Map任務(wù)，且這M個Map任務(wù)是分布于N臺計算機上并行運行的。Reduce任務(wù)的數(shù)量R則可由用戶指定。并行計算的實現(xiàn)-5PartitionPartition（劃分）是可選配置，主要作用是在有多個Reduce的情況下，指定Map的結(jié)果由某一個Reduce處理，每一個Reduce都會有單獨的輸出文件。Partition會把Map任務(wù)輸出的中間結(jié)果按鍵的范圍劃分成R份（R是預(yù)先定義的Reduce任務(wù)的個數(shù)），劃分時通常使用Hash函數(shù)，這樣可以保證某一范圍內(nèi)的鍵一定是由一個Reduce任務(wù)來處理，簡化Reduce的過程。并行計算的實現(xiàn)-6Combine在Partition之前，還可以對中間結(jié)果先進(jìn)行Combine（合并），即將中間結(jié)果中有相同鍵的鍵值對合并成一對。Combiner是可選擇的，它的主要作用是在每一個Map執(zhí)行完分析以后，在本地優(yōu)先做Reduce的工作，減少中間結(jié)果中鍵值對的數(shù)目，從而減少在Reduce過程中的數(shù)據(jù)傳輸量。Combine的過程與Reduce的過程類似，但Combine是作為Map任務(wù)的一部分，在執(zhí)行完Map函數(shù)后緊接著執(zhí)行的。并行計算的實現(xiàn)-7Reduce任務(wù)從Map任務(wù)節(jié)點獲取中間結(jié)果Map任務(wù)的中間結(jié)果在完成Combine和Partition之后，以文件形式保存于本地磁盤。中間結(jié)果文件的位置會被通知給主控JobTracker，由JobTracker再通知Reduce任務(wù)到哪個DataNode上去取中間結(jié)果。注意：所有Map任務(wù)產(chǎn)生的中間結(jié)果均按其鍵的范圍使用同一個Hash函數(shù)劃分成了R份，R個Reduce任務(wù)各自負(fù)責(zé)一段鍵區(qū)間。每個Reduce需要從多個Map任務(wù)節(jié)點獲取落在其負(fù)責(zé)的鍵區(qū)間內(nèi)的中間結(jié)果，然后執(zhí)行Reduce函數(shù)，形成一個最終的結(jié)果文件。實例分析：WordCount如果需要統(tǒng)計過去10年計算機論文中出現(xiàn)次數(shù)最多的幾個單詞，以分析當(dāng)前的熱點研究議題是什么，那將論文樣本收集完畢之后，接下來該怎么辦呢？如同Java中的“HelloWorld”經(jīng)典程序一樣，WordCount是MapReduce的入門程序，該程序要求計算出文件中各個單詞的頻數(shù)，并將輸出結(jié)果按照單詞的字母順序進(jìn)行排序，每個單詞和其頻數(shù)占一行，且單詞和頻數(shù)之間有間隔。例如，輸入一個文件，其內(nèi)容如下。helloworldhellohadoophellomapreduce其符合要求的輸出結(jié)果為。hadoop1hello3mapreduce1world1。上面這個應(yīng)用實例，就是將文件內(nèi)容切分成單詞，然后將所有相同的單詞聚集到一起，最后計算單詞出現(xiàn)的次數(shù)并輸出計算結(jié)果。每個拿到原始數(shù)據(jù)的機器只要將輸入數(shù)據(jù)切分成單詞就可以了，因此，可以在Map階段完成單詞切分的任務(wù)；另外，可將相同的單詞交給一臺機器來計算頻數(shù)，然后輸出最終結(jié)果，該任務(wù)可由Reduce階段完成；至于將中間結(jié)果根據(jù)不同單詞進(jìn)行分組，然后再發(fā)送給Reduce機器的任務(wù)，則可由MapReduce中的Shuffle階段完成。至此，上述實例的MapReduce程序就設(shè)計完成了：Map階段完成由輸入數(shù)據(jù)到單詞切分的工作；Shuffle階段完成相同單詞的聚集和分發(fā)工作；Reduce階段則完成接收所有單詞并計算其頻數(shù)的工作。WordCount類程序源代碼-1publicclassWordCount{publicstaticvoidmain(String[]args)throwsException{Configurationconf=newConfiguration();String[]otherArgs=newGenericOptionsParser(conf,args).getRemainingArgs();if(otherArgs.length<2){System.err.println("Usage:wordcount<in>[<in>...]<out>");System.exit(2);}Jobjob=Job.getInstance(conf,"wordcount");job.setJarByClass(WordCount.class);job.setMapperClass(TokenizerMapper.class);job.setCombinerClass(IntSumReducer.class);job.setReducerClass(IntSumReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);for(inti=0;i<otherArgs.length-1;i++){FileInputFormat.addInputPath(job,newPath(otherArgs[i]));}FileOutputFormat.setOutputPath(job,newPath(otherArgs[(otherArgs.length-1)]));

System.exit(job.waitForCompletion(true)?0:1);}

WordCount類程序源代碼-2publicstaticclassIntSumReducerextendsReducer<Text,IntWritable,Text,IntWritable>{privateIntWritableresult=newIntWritable();

publicvoidreduce(Textkey,Iterable<IntWritable>values,Reducer<Text,IntWritable,Text,IntWritable>.Contextcontext)throwsIOException,InterruptedException{intsum=0;for(IntWritableval:values){sum+=val.get();}this.result.set(sum);context.write(key,this.result);}}

publicstaticclassTokenizerMapperextendsMapper<Object,Text,Text,IntWritable>{privatestaticfinalIntWritableone=newIntWritable(1);privateTextword=newText();

publicvoidmap(Objectkey,Textvalue,Mapper<Object,Text,Text,IntWritable>.Contextcontext)throwsIOException,InterruptedException{StringTokenizeritr=newStringTokenizer(value.toString());while(itr.hasMoreTokens()){this.word.set(itr.nextToken());context.write(this.word,one);}}}}WordCount執(zhí)行過程整個過程如圖（Map階段與Reduce階段）使用Hadoop運行程序-1執(zhí)行如下命令，在HDFS中，創(chuàng)建input文件目錄。$hadoopfs-mkdir/input在Hadoop目錄下，找到文件LICENSE.txt，執(zhí)行如下命令，將其放到HDFS的input目錄下。$hadoopfs-putLICENSE.txt/input執(zhí)行如下命令，查看是否成功上傳文件。$hadoopfs-lshdfs:/inputFound1items-rw-r--r--3hadoopsupergroup154292016-09-0522:44hdfs:///input/LICENSE.txt執(zhí)行如下命令，運行WordCount程序。$hadoopjarshare/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.0.jarwordcount/input/output使用Hadoop運行程序-2程序運行的結(jié)果以文件的形式存放在HDFS系統(tǒng)的output目錄，地址如下。hadoop@ubuntu:~/soft/hadoop-2.7.0$hadoopfs-gethdfs:/output將運行結(jié)果文件下載到本地，查看統(tǒng)計結(jié)果，如圖MapReduce新框架YARN-1YARN帶來了巨大的改變，它改變了Hadoop計算組件（MapReduce）切分和重新組成處理任務(wù)的方式，因為YARN把MapReduce的追蹤組件切分成兩個不同的部分：資源管理器和應(yīng)用調(diào)度，有助于更加輕松地同時運行MapReduce或Storm這樣的任務(wù)以及HBase等服務(wù)。YARN框架的原理及運作機制：重構(gòu)后新的HadoopMapReduce框架被命名為MapReduceV2或者YARN，其架構(gòu)如圖MapReduce新框架YARN-2此次重構(gòu)的基本思想是：將JobTracker的兩個主要功能——資源管理和任務(wù)調(diào)度/監(jiān)控——分離成單獨的組件。新的ResourceManager全局管理所有應(yīng)用程序計算資源的分配，而每一個應(yīng)用的ApplicationMaster負(fù)責(zé)相應(yīng)的調(diào)度和協(xié)調(diào)。實際上，每一個應(yīng)用的ApplicationMaster都是一個詳細(xì)的框架庫，它結(jié)合從ResourceManager獲得的資源，與NodeManager協(xié)同工作來執(zhí)行并監(jiān)控任務(wù)。某種意義上講，ResourceManager就是一個純粹的調(diào)度器，它在執(zhí)行過程中不對應(yīng)用進(jìn)行監(jiān)控和狀態(tài)跟蹤，同樣，它也不能重啟因應(yīng)用失敗或硬件錯誤而運行失敗的任務(wù)。ResourceManager是基于應(yīng)用程序?qū)Y源的需求進(jìn)行調(diào)度的，資源包括內(nèi)存、CPU、磁盤、網(wǎng)絡(luò)等等，每一個應(yīng)用程序需要不同類型的資源，因此就需要不同的容器?？梢钥闯?，這與原來的MapReduce固定類型的資源使用模型有顯著區(qū)別，原來的模型會給集群的使用帶來負(fù)面影響。ResourceManager提供一個調(diào)度策略插件，它負(fù)責(zé)將集群資源分配給多個隊列和應(yīng)用程序，調(diào)度插件可以基于現(xiàn)有的能力調(diào)度資源。NodeManager是每一臺機器框架的代理，是執(zhí)行應(yīng)用程序的容器，它監(jiān)控應(yīng)用程序的資源使用情況（CPU、內(nèi)存、硬盤、網(wǎng)絡(luò)等）并向調(diào)度器匯報。新舊HadoopMapReduce框架比對新的YARN框架相對于舊的MapRduce框架而言，其配置文件、啟停腳本及全局變量等都發(fā)生了一些變化，原框架中核心的JobTracker和TaskTracker不見了，取而代之的是ResourceManager、ApplicationMaster與NodeManager三個部分，三者的具體功能如下：ResourceManager：ResourceManager是一個中心的服務(wù)，它的工作是調(diào)度、啟動每一個Job所屬的ApplicationMaster，并且監(jiān)控ApplicationMaster的存在情況。ResourceManager負(fù)責(zé)作業(yè)與資源的調(diào)度，它接收J(rèn)obSubmitter提交的作業(yè)，按照作業(yè)的上下文（Context）信息以及從NodeManager收集來的狀態(tài)信息，啟動調(diào)度過程，并分配一個Container作為ApplicationMaster。ApplicationMaster：負(fù)責(zé)一個Job生命周期內(nèi)的所有工作，類似舊MapReduce框架中的JobTracker。NodeManager：功能比較專一，即負(fù)責(zé)Container狀態(tài)的維護(hù)，并與ResourceManager保持通信。YARN框架的優(yōu)勢大大減少了JobTracker（也就是現(xiàn)在的ResourceManager）的資源消耗，并且使監(jiān)測每一個Job子任務(wù)的程序分布式部署，使程序更安全、