1 概述

該瞅瞅MapReduce的內(nèi)部運(yùn)行原理了，以前只知道個皮毛，再不搞搞，不然怎么死的都不曉得。下文會以2.4版本中的WordCount這個經(jīng)典例子作為分析的切入點(diǎn)，一步步來看里面到底是個什么情況。

2 為什么要使用MapReduce

Map/Reduce，是一種模式，適合解決并行計算的問題，比如TopN、貝葉斯分類等。注意，是并行計算，而非迭代計算，像涉及到層次聚類的問題就不太適合了。

從名字可以看出，這種模式有兩個步驟，Map和Reduce。Map即數(shù)據(jù)的映射，用于把一組鍵值對映射成另一組新的鍵值對，而Reduce這個東東，以Map階段的輸出結(jié)果作為輸入，對數(shù)據(jù)做化簡、合并等操作。

而MapReduce是Hadoop生態(tài)系統(tǒng)中基于底層HDFS的一個計算框架，它的上層又可以是Hive、Pig等數(shù)據(jù)倉庫框架，也可以是Mahout這樣的數(shù)據(jù)挖掘工具。由于MapReduce依賴于HDFS，其運(yùn)算過程中的數(shù)據(jù)等會保存到HDFS上，把對數(shù)據(jù)集的計算分發(fā)給各個節(jié)點(diǎn)，并將結(jié)果進(jìn)行匯總，再加上各種狀態(tài)匯報、心跳匯報等，其只適合做離線計算。和實(shí)時計算框架Storm、Spark等相比，速度上沒有優(yōu)勢。舊的Hadoop生態(tài)幾乎是以MapReduce為核心的，但是慢慢的發(fā)展，其擴(kuò)展性差、資源利用率低、可靠性等問題都越來越讓人覺得不爽，于是才產(chǎn)生了Yarn這個新的東東，并且二代版的Hadoop生態(tài)都是以Yarn為核心。Storm、Spark等都可以基于Yarn使用。

3 怎么運(yùn)行MapReduce

明白了哪些地方可以使用這個牛叉的MapReduce框架，那該怎么用呢？Hadoop的MapReduce源碼給我們提供了范例，在其hadoop-mapreduce-examples子工程中包含了MapReduce的Java版例子。在寫完類似的代碼后，打包成jar，在HDFS的客戶端運(yùn)行：

bin/hadoop jar mapreduce_examples.jar mainClass args

即可。當(dāng)然，也可以在IDE(如Eclipse)中，進(jìn)行遠(yuǎn)程運(yùn)行、調(diào)試程序。

至于，HadoopStreaming方式，網(wǎng)上有很多。我們這里只討論Java的實(shí)現(xiàn)。

4 如何編寫MapReduce程序

??? 如前文所說，MapReduce中有Map和Reduce，在實(shí)現(xiàn)MapReduce的過程中，主要分為這兩個階段，分別以兩類函數(shù)進(jìn)行展現(xiàn)，一個是map函數(shù)，一個是reduce函數(shù)。map函數(shù)的參數(shù)是一個<key,value>鍵值對，其輸出結(jié)果也是鍵值對，reduce函數(shù)以map的輸出作為輸入進(jìn)行處理。

4.1 代碼構(gòu)成

??? 實(shí)際的代碼中，需要三個元素，分別是Map、Reduce、運(yùn)行任務(wù)的代碼。這里的Map類是繼承了org.apache.hadoop.mapreduce.Mapper，并實(shí)現(xiàn)其中的map方法；而Reduce類是繼承了org.apache.hadoop.mapreduce.Reducer，實(shí)現(xiàn)其中的reduce方法。至于運(yùn)行任務(wù)的代碼，就是我們程序的入口。

??? 下面是Hadoop提供的WordCount源碼。

1 /** 2 * Licensed to the Apache Software Foundation (ASF) under one 3 * or more contributor license agreements. See the NOTICE file 4 * distributed with this work for additional information 5 * regarding copyright ownership. The ASF licenses this file 6 * to you under the Apache License, Version 2.0 (the 7 * "License"); you may not use this file except in compliance 8 * with the License. You may obtain a copy of the License at 9 * 10 * http://www.apache.org/licenses/LICENSE-2.0 11 * 12 * Unless required by applicable law or agreed to in writing, software 13 * distributed under the License is distributed on an "AS IS" BASIS, 14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15 * See the License for the specific language governing permissions and 16 * limitations under the License. 17 */ 18 package org.apache.hadoop.examples; 19 20 import java.io.IOException; 21 import java.util.StringTokenizer; 22 23 import org.apache.hadoop.conf.Configuration; 24 import org.apache.hadoop.fs.Path; 25 import org.apache.hadoop.io.IntWritable; 26 import org.apache.hadoop.io.Text; 27 import org.apache.hadoop.mapreduce.Job; 28 import org.apache.hadoop.mapreduce.Mapper; 29 import org.apache.hadoop.mapreduce.Reducer; 30 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 31 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 32 import org.apache.hadoop.util.GenericOptionsParser; 33 34 public class WordCount { 35 36 public static class TokenizerMapper 37 extends Mapper<Object, Text, Text, IntWritable>{ 38 39 private final static IntWritable one = new IntWritable(1); 40 private Text word = new Text(); 41 42 public void map(Object key, Text value, Context context 43 ) throws IOException, InterruptedException { 44 StringTokenizer itr = new StringTokenizer(value.toString()); 45 while (itr.hasMoreTokens()) { 46 word.set(itr.nextToken()); 47 context.write(word, one); 48 } 49 } 50 } 51 52 public static class IntSumReducer 53 extends Reducer<Text,IntWritable,Text,IntWritable> { 54 private IntWritable result = new IntWritable(); 55 56 public void reduce(Text key, Iterable<IntWritable> values, 57 Context context 58 ) throws IOException, InterruptedException { 59 int sum = 0; 60 for (IntWritable val : values) { 61 sum += val.get(); 62 } 63 result.set(sum); 64 context.write(key, result); 65 } 66 } 67 68 public static void main(String[] args) throws Exception { 69 Configuration conf = new Configuration(); 70 String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); 71 if (otherArgs.length != 2) { 72 System.err.println("Usage: wordcount <in> <out>"); 73 System.exit(2); 74 } 75 Job job = new Job(conf, "word count"); 76 job.setJarByClass(WordCount.class); 77 job.setMapperClass(TokenizerMapper.class); 78 job.setCombinerClass(IntSumReducer.class); 79 job.setReducerClass(IntSumReducer.class); 80 job.setOutputKeyClass(Text.class); 81 job.setOutputValueClass(IntWritable.class); 82 FileInputFormat.addInputPath(job, new Path(otherArgs[0])); 83 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); 84 System.exit(job.waitForCompletion(true) ? 0 : 1); 85 } 86 } View Code

?

4.2 入口類

4.2.1 參數(shù)獲取

首先定義配置文件類Configuration，此類是Hadoop各個模塊的公共使用類，用于加載類路徑下的各種配置文件，讀寫其中的配置選項(xiàng)。

??? 第二步中，用到了GenericOptionsParser類，其目的是將命令行中參數(shù)自動設(shè)置到變量conf中。

??? GenericOptionsParser的構(gòu)造方法進(jìn)去之后，會進(jìn)行到parseGeneralOptions，對傳入的參數(shù)進(jìn)行解析：

1 private void parseGeneralOptions(Options opts, Configuration conf, 2 3 String[] args) throws IOException { 4 5 opts = buildGeneralOptions(opts); 6 7 CommandLineParser parser = new GnuParser(); 8 9 try { 10 11 commandLine = parser.parse(opts, preProcessForWindows(args), true); 12 13 processGeneralOptions(conf, commandLine); 14 15 } catch(ParseException e) { 16 17 LOG.warn("options parsing failed: "+e.getMessage()); 18 19 20 21 HelpFormatter formatter = new HelpFormatter(); 22 23 formatter.printHelp("general options are: ", opts); 24 25 } 26 27 }

?

? ?而getRemainingArgs方法會獲得傳入的參數(shù)，接著在main方法中會進(jìn)行判斷參數(shù)的個數(shù)，由于此處是WordCount計算，只需要傳入文件的輸入路徑和輸出路徑即可，因此參數(shù)的個數(shù)為2，否則將退出：

1 if (otherArgs.length != 2) { 2 3 System.err.println("Usage: wordcount <in> <out>"); 4 5 System.exit(2); 6 7 }

?

如果在代碼運(yùn)行的時候傳入其他的參數(shù)，比如指定reduce的個數(shù)，可以根據(jù)GenericOptionsParser的命令行格式這么寫：

bin/hadoop jar MyJob.jar com.xxx.MyJobDriver -Dmapred.reduce.tasks=5

其規(guī)則是-D加MapReduce的配置選項(xiàng)，當(dāng)然還支持-fs等其他參數(shù)傳入。當(dāng)然，默認(rèn)情況下Reduce的數(shù)目為1，Map的數(shù)目也為1。

4.2.2 Job定義

?? 定義Job對象，其構(gòu)造方法為：

1 public Job(Configuration conf, String jobName) throws IOException { 2 3 this(conf); 4 5 setJobName(jobName); 6 7 }

?

可見，傳入的"word count"就是Job的名字。而conf被傳遞給了JobConf進(jìn)行環(huán)境變量的獲取：

1 public JobConf(Configuration conf) { 2 3 super(conf); 6 7 if (conf instanceof JobConf) { 8 9 JobConf that = (JobConf)conf; 10 11 credentials = that.credentials; 12 13 } 14 checkAndWarnDeprecation(); 19 }

?

??? Job已經(jīng)實(shí)例化了，下面就得給這個Job加點(diǎn)佐料才能讓它按照我們的要求運(yùn)行。于是依次給Job添加啟動Jar包、設(shè)置Mapper類、設(shè)置合并類、設(shè)置Reducer類、設(shè)置輸出鍵類型、設(shè)置輸出值的類型。

??? 這里有必要說下設(shè)置Jar包的這個方法setJarByClass：

1 public void setJarByClass(Class<?> cls) { 2 3 ensureState(JobState.DEFINE); 4 5 conf.setJarByClass(cls); 6 7 }

?

它會首先判斷當(dāng)前Job的狀態(tài)是否是運(yùn)行中，接著通過class找到其所屬的jar文件，將jar路徑賦值給mapreduce.job.jar屬性。至于尋找jar文件的方法，則是通過classloader獲取類路徑下的資源文件，進(jìn)行循環(huán)遍歷。具體實(shí)現(xiàn)見ClassUtil類中的findContainingJar方法。

??? 搞完了上面的東西，緊接著就會給mapreduce.input.fileinputformat.inputdir參數(shù)賦值，這是Job的輸入路徑，還有mapreduce.input.fileinputformat.inputdir，這是Job的輸出路徑。具體的位置，就是我們前面main中傳入的Args。

4.2.3 Job提交

??? 萬事俱備，那就運(yùn)行吧。

??? 這里調(diào)用的方法如下：

1 public boolean waitForCompletion(boolean verbose 2 3 ) throws IOException, InterruptedException, 4 5 ClassNotFoundException { 6 7 if (state == JobState.DEFINE) { 8 9 submit(); 10 11 } 12 13 if (verbose) { 14 15 monitorAndPrintJob(); 16 17 } else { 18 19 // get the completion poll interval from the client. 20 21 int completionPollIntervalMillis = 22 23 Job.getCompletionPollInterval(cluster.getConf()); 24 25 while (!isComplete()) { 26 27 try { 28 29 Thread.sleep(completionPollIntervalMillis); 30 31 } catch (InterruptedException ie) { 32 33 } 34 35 } 36 37 } 38 39 return isSuccessful(); 40 41 }

?

至于方法的參數(shù)verbose，如果想在控制臺打印當(dāng)前的進(jìn)度，則設(shè)置為true。

?? 至于submit方法，如果當(dāng)前在HDFS的配置文件中配置了mapreduce.framework.name屬性為“yarn”的話，會創(chuàng)建一個YARNRunner對象來進(jìn)行任務(wù)的提交。其構(gòu)造方法如下：

1 public YARNRunner(Configuration conf, ResourceMgrDelegate resMgrDelegate, 2 3 ClientCache clientCache) { 4 5 this.conf = conf; 6 7 try { 8 9 this.resMgrDelegate = resMgrDelegate; 10 11 this.clientCache = clientCache; 12 13 this.defaultFileContext = FileContext.getFileContext(this.conf); 14 15 } catch (UnsupportedFileSystemException ufe) { 16 17 throw new RuntimeException("Error in instantiating YarnClient", ufe); 18 19 } 20 21 }

?

其中，ResourceMgrDelegate實(shí)際上ResourceManager的代理類，其實(shí)現(xiàn)了YarnClient接口，通過ApplicationClientProtocol代理直接向RM提交Job，殺死Job，查看Job運(yùn)行狀態(tài)等操作。同時，在ResourceMgrDelegate類中會通過YarnConfiguration來讀取yarn-site.xml、core-site.xml等配置文件中的配置屬性。

?? 下面就到了客戶端最關(guān)鍵的時刻了，提交Job到集群運(yùn)行。具體實(shí)現(xiàn)類是JobSubmitter類中的submitJobInternal方法。這個牛氣哄哄的方法寫了100多行，還不算其幾十行的注釋。我們看它干了點(diǎn)啥。

Step1:

檢查job的輸出路徑是否存在，如果存在則拋出異常。

Step2:

初始化用于存放Job相關(guān)資源的路徑。注意此路徑的構(gòu)造方式為：

1 conf.get(MRJobConfig.MR_AM_STAGING_DIR, 2 3 MRJobConfig.DEFAULT_MR_AM_STAGING_DIR) 4 5 + Path.SEPARATOR + user 6 7 + Path.SEPARATOR + STAGING_CONSTANT

?

其中，MRJobConfig.DEFAULT_MR_AM_STAGING_DIR為“/tmp/hadoop-yarn/staging”，STAGING_CONSTANT為".staging"。

Step3:

設(shè)置客戶端的host屬性：mapreduce.job.submithostname和mapreduce.job.submithostaddress。

Step4:

通過RPC，向Yarn的ResourceManager申請JobID對象。

Step5:

從HDFS的NameNode獲取驗(yàn)證用的Token，并將其放入緩存。

Step6:

將作業(yè)文件上傳到HDFS，這里如果我們前面沒有對Job命名的話，默認(rèn)的名稱就會在這里設(shè)置成jar的名字。并且，作業(yè)默認(rèn)的副本數(shù)是10，如果屬性mapreduce.client.submit.file.replication沒有被設(shè)置的話。

Step7:

文件上傳到HDFS之后，還要被DistributedCache進(jìn)行緩存起來。這是因?yàn)橛嬎愎?jié)點(diǎn)收到該作業(yè)的第一個任務(wù)后，就會有DistributedCache自動將作業(yè)文件Cache到節(jié)點(diǎn)本地目錄下，并且會對壓縮文件進(jìn)行解壓，如：.zip，.jar，.tar等等，然后開始任務(wù)。

最后，對于同一個計算節(jié)點(diǎn)接下來收到的任務(wù)，DistributedCache不會重復(fù)去下載作業(yè)文件，而是直接運(yùn)行任務(wù)。如果一個作業(yè)的任務(wù)數(shù)很多，這種設(shè)計避免了在同一個節(jié)點(diǎn)上對用一個job的文件會下載多次，大大提高了任務(wù)運(yùn)行的效率。

Step8:

對每個輸入文件進(jìn)行split劃分。注意這只是個邏輯的劃分，不是物理的。因?yàn)榇颂幨禽斎胛募?#xff0c;因此執(zhí)行的是FileInputFormat類中的getSplits方法。只有非壓縮的文件和幾種特定壓縮方式壓縮后的文件才分片。分片的大小由如下幾個參數(shù)決定：mapreduce.input.fileinputformat.split.maxsize、mapreduce.input.fileinputformat.split.minsize、文件的塊大小。

具體計算方式為：

Math.max(minSize, Math.min(maxSize, blockSize))

分片的大小有可能比默認(rèn)塊大小64M要大，當(dāng)然也有可能小于它，默認(rèn)情況下分片大小為當(dāng)前HDFS的塊大小，64M。

?? 接下來就該正兒八經(jīng)的獲取分片詳情了。代碼如下：

1 long bytesRemaining = length; 2 3 while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) { 4 5 int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining); 6 7 splits.add(makeSplit(path, length-bytesRemaining, splitSize, 9 blkLocations[blkIndex].getHosts())); 10 11 bytesRemaining -= splitSize; 13 } 15 16 if (bytesRemaining != 0) { 18 int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining); 19 20 splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining, 22 blkLocations[blkIndex].getHosts())); 23 24 }

?

Step8.1：

?? 將bytesRemaining(剩余未分片字節(jié)數(shù))設(shè)置為整個文件的長度。

Step8.2：

如果bytesRemaining超過分片大小splitSize一定量才會將文件分成多個InputSplit，SPLIT_SLOP(默認(rèn)1.1)。接著就會執(zhí)行如下方法獲取block的索引，其中第二個參數(shù)是這個block在整個文件中的偏移量，在循環(huán)中會從0越來越大：

1 protected int getBlockIndex(BlockLocation[] blkLocations, long offset) { 4 for (int i = 0 ; i < blkLocations.length; i++) { 5 // is the offset inside this block? 6 if ((blkLocations[i].getOffset() <= offset) && 7 (offset < blkLocations[i].getOffset() + blkLocations[i].getLength())){ 8 return i; 9 } 10 } 11 12 BlockLocation last = blkLocations[blkLocations.length -1]; 13 long fileLength = last.getOffset() + last.getLength() -1; 14 throw new IllegalArgumentException("Offset " + offset + " is outside of file (0.." + fileLength + ")"); 17 }

?

將符合條件的塊的索引對應(yīng)的block信息的主機(jī)節(jié)點(diǎn)以及文件的路徑名、開始的偏移量、分片大小splitSize封裝到一個InputSplit中加入List<InputSplit> splits。

Step8.3：

bytesRemaining -= splitSize修改剩余字節(jié)大小。剩余如果bytesRemaining還不為0，表示還有未分配的數(shù)據(jù)，將剩余的數(shù)據(jù)及最后一個block加入splits。

Step8.4

如果不允許分割isSplitable==false，則將第一個block、文件目錄、開始位置為0，長度為整個文件的長度封裝到一個InputSplit，加入splits中；如果文件的長度==0，則splits.add(new FileSplit(path, 0, length, new String[0]))沒有block，并且初始和長度都為0；

Step8.5

將輸入目錄下文件的個數(shù)賦值給?"mapreduce.input.num.files"，方便以后校對，返回分片信息splits。

　　這就是getSplits獲取分片的過程。當(dāng)使用基于FileInputFormat實(shí)現(xiàn)InputFormat時，為了提高M(jìn)apTask的數(shù)據(jù)本地性，應(yīng)盡量使InputSplit大小與block大小相同。

?　如果分片大小超過bolck大小，但是InputSplit中的封裝了單個block的所在主機(jī)信息啊，這樣能讀取多個bolck數(shù)據(jù)嗎？

比如當(dāng)前文件很大，1G，我們設(shè)置的最小分片是100M，最大是200M，當(dāng)前塊大小為64M，經(jīng)過計算后的實(shí)際分片大小是100M，這個時候第二個分片中存放的也只是一個block的host信息。需要注意的是split是邏輯分片，不是物理分片，當(dāng)Map任務(wù)需要的數(shù)據(jù)本地性發(fā)揮作用時，會從本機(jī)的block開始讀取，超過這個block的部分可能不在本機(jī)，這就需要從別的DataNode拉數(shù)據(jù)過來，因?yàn)閷?shí)際獲取數(shù)據(jù)是一個輸入流，這個輸入流面向的是整個文件，不受split的影響，split的大小越大可能需要從別的節(jié)點(diǎn)拉的數(shù)據(jù)越多，從從而效率也會越慢，拉數(shù)據(jù)的多少是由getSplits方法中的splitSize決定的。所以為了更有效率，分片的大小盡量保持在一個block大小吧。

Step9:

將split信息和SplitMetaInfo都寫入HDFS中。使用方法：

1 JobSplitWriter.createSplitFiles(jobSubmitDir, conf, jobSubmitDir.getFileSystem(conf), array);

?

Step10:

對Map數(shù)目設(shè)置，上面獲得到的split的個數(shù)就是實(shí)際的Map任務(wù)的數(shù)目。

Step11:

相關(guān)配置寫入到j(luò)ob.xml中：

1 jobCopy.writeXml(out);

?

Step12:

通過如下代碼正式提交Job到Y(jié)arn：

1 status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());

?

?? 這里就涉及到Y(jié)arnClient和RresourceManager的RPC通信了。包括獲取applicationId、進(jìn)行狀態(tài)檢查、網(wǎng)絡(luò)通信等。

Step13:

上面通過RPC的調(diào)用，最后會返回一個JobStatus對象，它的toString方法可以在JobClient端打印運(yùn)行的相關(guān)日志信息。

4.2.4 另一種運(yùn)行方式

?? 提交MapReduce任務(wù)的方式除了上述源碼中給出的之外，還可以使用ToolRunner方式。具體方式為：

1 ToolRunner.run(new Configuration(),new WordCount(), args);

?

至此，我們的MapReduce的啟動類要做的事情已經(jīng)分析完了。

?

-------------------------------------------------------------------------------

如果您看了本篇博客,覺得對您有所收獲，請點(diǎn)擊右下角的?[推薦]

如果您想轉(zhuǎn)載本博客，請注明出處

如果您對本文有意見或者建議，歡迎留言

感謝您的閱讀，請關(guān)注我的后續(xù)博客

轉(zhuǎn)載于:https://www.cnblogs.com/Scott007/p/3836687.html

總結(jié)

以上是生活随笔為你收集整理的Mapreduce执行过程分析(基于Hadoop2.4)——(一)的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。