知识点
相关文章
更多最近更新
更多Mapreduce执行过程分析(基于Hadoop2.4)——(一)
2019-03-02 23:43|来源: 网路
1 概述
该瞅瞅MapReduce的内部运行原理了,以前只知道个皮毛,再不搞搞,不然怎么死的都不晓得。下文会以2.4版本中的WordCount这个经典例子作为分析的切入点,一步步来看里面到底是个什么情况。
2 为什么要使用MapReduce
Map/Reduce,是一种模式,适合解决并行计算的问题,比如TopN、贝叶斯分类等。注意,是并行计算,而非迭代计算,像涉及到层次聚类的问题就不太适合了。
从名字可以看出,这种模式有两个步骤,Map和Reduce。Map即数据的映射,用于把一组键值对映射成另一组新的键值对,而Reduce这个东东,以Map阶段的输出结果作为输入,对数据做化简、合并等操作。
而MapReduce是Hadoop生态系统中基于底层HDFS的一个计算框架,它的上层又可以是Hive、Pig等数据仓库框架,也可以是Mahout这样的数据挖掘工具。由于MapReduce依赖于HDFS,其运算过程中的数据等会保存到HDFS上,把对数据集的计算分发给各个节点,并将结果进行汇总,再加上各种状态汇报、心跳汇报等,其只适合做离线计算。和实时计算框架Storm、Spark等相比,速度上没有优势。旧的Hadoop生态几乎是以MapReduce为核心的,但是慢慢的发展,其扩展性差、资源利用率低、可靠性等问题都越来越让人觉得不爽,于是才产生了Yarn这个新的东东,并且二代版的Hadoop生态都是以Yarn为核心。Storm、Spark等都可以基于Yarn使用。
3 怎么运行MapReduce
明白了哪些地方可以使用这个牛叉的MapReduce框架,那该怎么用呢?Hadoop的MapReduce源码给我们提供了范例,在其hadoop-mapreduce-examples子工程中包含了MapReduce的Java版例子。在写完类似的代码后,打包成jar,在HDFS的客户端运行:
bin/hadoop jar mapreduce_examples.jar mainClass args
即可。当然,也可以在IDE(如Eclipse)中,进行远程运行、调试程序。
至于,HadoopStreaming方式,网上有很多。我们这里只讨论Java的实现。
4 如何编写MapReduce程序
如前文所说,MapReduce中有Map和Reduce,在实现MapReduce的过程中,主要分为这两个阶段,分别以两类函数进行展现,一个是map函数,一个是reduce函数。map函数的参数是一个<key,value>键值对,其输出结果也是键值对,reduce函数以map的输出作为输入进行处理。
4.1 代码构成
实际的代码中,需要三个元素,分别是Map、Reduce、运行任务的代码。这里的Map类是继承了org.apache.hadoop.mapreduce.Mapper,并实现其中的map方法;而Reduce类是继承了org.apache.hadoop.mapreduce.Reducer,实现其中的reduce方法。至于运行任务的代码,就是我们程序的入口。
下面是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 }
4.2 入口类
4.2.1 参数获取
首先定义配置文件类Configuration,此类是Hadoop各个模块的公共使用类,用于加载类路径下的各种配置文件,读写其中的配置选项。
第二步中,用到了GenericOptionsParser类,其目的是将命令行中参数自动设置到变量conf中。
GenericOptionsParser的构造方法进去之后,会进行到parseGeneralOptions,对传入的参数进行解析:
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方法会获得传入的参数,接着在main方法中会进行判断参数的个数,由于此处是WordCount计算,只需要传入文件的输入路径和输出路径即可,因此参数的个数为2,否则将退出:
1 if (otherArgs.length != 2) { 2 3 System.err.println("Usage: wordcount <in> <out>"); 4 5 System.exit(2); 6 7 }
如果在代码运行的时候传入其他的参数,比如指定reduce的个数,可以根据GenericOptionsParser的命令行格式这么写:
bin/hadoop jar MyJob.jar com.xxx.MyJobDriver -Dmapred.reduce.tasks=5
其规则是-D加MapReduce的配置选项,当然还支持-fs等其他参数传入。当然,默认情况下Reduce的数目为1,Map的数目也为1。
4.2.2 Job定义
定义Job对象,其构造方法为:
1 public Job(Configuration conf, String jobName) throws IOException { 2 3 this(conf); 4 5 setJobName(jobName); 6 7 }
可见,传入的"word count"就是Job的名字。而conf被传递给了JobConf进行环境变量的获取:
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已经实例化了,下面就得给这个Job加点佐料才能让它按照我们的要求运行。于是依次给Job添加启动Jar包、设置Mapper类、设置合并类、设置Reducer类、设置输出键类型、设置输出值的类型。
这里有必要说下设置Jar包的这个方法setJarByClass:
1 public void setJarByClass(Class<?> cls) { 2 3 ensureState(JobState.DEFINE); 4 5 conf.setJarByClass(cls); 6 7 }
它会首先判断当前Job的状态是否是运行中,接着通过class找到其所属的jar文件,将jar路径赋值给mapreduce.job.jar属性。至于寻找jar文件的方法,则是通过classloader获取类路径下的资源文件,进行循环遍历。具体实现见ClassUtil类中的findContainingJar方法。
搞完了上面的东西,紧接着就会给mapreduce.input.fileinputformat.inputdir参数赋值,这是Job的输入路径,还有mapreduce.input.fileinputformat.inputdir,这是Job的输出路径。具体的位置,就是我们前面main中传入的Args。
4.2.3 Job提交
万事俱备,那就运行吧。
这里调用的方法如下:
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 }
至于方法的参数verbose,如果想在控制台打印当前的进度,则设置为true。
至于submit方法,如果当前在HDFS的配置文件中配置了mapreduce.framework.name属性为“yarn”的话,会创建一个YARNRunner对象来进行任务的提交。其构造方法如下:
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实际上ResourceManager的代理类,其实现了YarnClient接口,通过ApplicationClientProtocol代理直接向RM提交Job,杀死Job,查看Job运行状态等操作。同时,在ResourceMgrDelegate类中会通过YarnConfiguration来读取yarn-site.xml、core-site.xml等配置文件中的配置属性。
下面就到了客户端最关键的时刻了,提交Job到集群运行。具体实现类是JobSubmitter类中的submitJobInternal方法。这个牛气哄哄的方法写了100多行,还不算其几十行的注释。我们看它干了点啥。
Step1:
检查job的输出路径是否存在,如果存在则抛出异常。
Step2:
初始化用于存放Job相关资源的路径。注意此路径的构造方式为:
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:
设置客户端的host属性:mapreduce.job.submithostname和mapreduce.job.submithostaddress。
Step4:
通过RPC,向Yarn的ResourceManager申请JobID对象。
Step5:
从HDFS的NameNode获取验证用的Token,并将其放入缓存。
Step6:
将作业文件上传到HDFS,这里如果我们前面没有对Job命名的话,默认的名称就会在这里设置成jar的名字。并且,作业默认的副本数是10,如果属性mapreduce.client.submit.file.replication没有被设置的话。
Step7:
文件上传到HDFS之后,还要被DistributedCache进行缓存起来。这是因为计算节点收到该作业的第一个任务后,就会有DistributedCache自动将作业文件Cache到节点本地目录下,并且会对压缩文件进行解压,如:.zip,.jar,.tar等等,然后开始任务。
最后,对于同一个计算节点接下来收到的任务,DistributedCache不会重复去下载作业文件,而是直接运行任务。如果一个作业的任务数很多,这种设计避免了在同一个节点上对用一个job的文件会下载多次,大大提高了任务运行的效率。
Step8:
对每个输入文件进行split划分。注意这只是个逻辑的划分,不是物理的。因为此处是输入文件,因此执行的是FileInputFormat类中的getSplits方法。只有非压缩的文件和几种特定压缩方式压缩后的文件才分片。分片的大小由如下几个参数决定:mapreduce.input.fileinputformat.split.maxsize、mapreduce.input.fileinputformat.split.minsize、文件的块大小。
具体计算方式为:
Math.max(minSize, Math.min(maxSize, blockSize))
分片的大小有可能比默认块大小64M要大,当然也有可能小于它,默认情况下分片大小为当前HDFS的块大小,64M。
接下来就该正儿八经的获取分片详情了。代码如下:
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(剩余未分片字节数)设置为整个文件的长度。
Step8.2:
如果bytesRemaining超过分片大小splitSize一定量才会将文件分成多个InputSplit,SPLIT_SLOP(默认1.1)。接着就会执行如下方法获取block的索引,其中第二个参数是这个block在整个文件中的偏移量,在循环中会从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 }
将符合条件的块的索引对应的block信息的主机节点以及文件的路径名、开始的偏移量、分片大小splitSize封装到一个InputSplit中加入List<InputSplit> splits。
Step8.3:
bytesRemaining -= splitSize修改剩余字节大小。剩余如果bytesRemaining还不为0,表示还有未分配的数据,将剩余的数据及最后一个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
将输入目录下文件的个数赋值给 "mapreduce.input.num.files",方便以后校对,返回分片信息splits。
这就是getSplits获取分片的过程。当使用基于FileInputFormat实现InputFormat时,为了提高MapTask的数据本地性,应尽量使InputSplit大小与block大小相同。
如果分片大小超过bolck大小,但是InputSplit中的封装了单个block的所在主机信息啊,这样能读取多个bolck数据吗?
比如当前文件很大,1G,我们设置的最小分片是100M,最大是200M,当前块大小为64M,经过计算后的实际分片大小是100M,这个时候第二个分片中存放的也只是一个block的host信息。需要注意的是split是逻辑分片,不是物理分片,当Map任务需要的数据本地性发挥作用时,会从本机的block开始读取,超过这个block的部分可能不在本机,这就需要从别的DataNode拉数据过来,因为实际获取数据是一个输入流,这个输入流面向的是整个文件,不受split的影响,split的大小越大可能需要从别的节点拉的数据越多,从从而效率也会越慢,拉数据的多少是由getSplits方法中的splitSize决定的。所以为了更有效率,分片的大小尽量保持在一个block大小吧。
Step9:
将split信息和SplitMetaInfo都写入HDFS中。使用方法:
1 JobSplitWriter.createSplitFiles(jobSubmitDir, conf, jobSubmitDir.getFileSystem(conf), array);
Step10:
对Map数目设置,上面获得到的split的个数就是实际的Map任务的数目。
Step11:
相关配置写入到job.xml中:
1 jobCopy.writeXml(out);
Step12:
通过如下代码正式提交Job到Yarn:
1 status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());
这里就涉及到YarnClient和RresourceManager的RPC通信了。包括获取applicationId、进行状态检查、网络通信等。
Step13:
上面通过RPC的调用,最后会返回一个JobStatus对象,它的toString方法可以在JobClient端打印运行的相关日志信息。
4.2.4 另一种运行方式
提交MapReduce任务的方式除了上述源码中给出的之外,还可以使用ToolRunner方式。具体方式为:
1 ToolRunner.run(new Configuration(),new WordCount(), args);
至此,我们的MapReduce的启动类要做的事情已经分析完了。
-------------------------------------------------------------------------------
如果您看了本篇博客,觉得对您有所收获,请点击右下角的 [推荐]
如果您想转载本博客,请注明出处
如果您对本文有意见或者建议,欢迎留言
感谢您的阅读,请关注我的后续博客
转自:http://www.cnblogs.com/Scott007/p/3836687
相关问答
更多-
如何在docker中部署hadoop2.4集群[2023-10-14]
Docker提供了一种完美隔离和打包Hadoop相关的应用程序的方法。该公司也在观察Slider框架和Docker如何能够一起合作来简化此类的部署工作。目前,Hortonworks的一个客户已经在考虑他们自己的数据平台利用HDP支持Docker的方法。他们利用Cloudbreak在云中的Docker容器中部署Hadoop,并计划把他们自己的数据应用程序制作成Docker镜像以便在YARN上运行。此外,很多用户也因为这种方式的敏捷性和兼容性,利用其进行未知环境的部署。 -
如何通过eclipse查看,阅读hadoop2.4源码[2023-05-03]
1.导入查看hadoop源码 (1)选择Existing Projects into Workspace (2)选择源码路径 (3)查看源码 这样我们就完成了全部的内容。 2.阅读hadoop源码 其中比较常用的: Open Call Hierarchy: 用Open Call Hierarchy可以查看方法的调用层次。如果想知道一个方法在别的什么地方被调用了,这个功能就很好用了,实际中也很常用,比如要重构一个方法时,想 知道他对其它什么地方可能有影响,就可以用这个功能。在方 ... -
在各个slave(datanode)上面有会有Map和Reduce执行代码。 在Job提交时,会打包该job的配置文件类文件,jar文件等,拷 贝到各个datanode上面,做本地执行的。
-
hadoop2.4版本中yarn的web管理界面不能查看作业状态!!!求助[2022-05-03]
no router to host...ssh方面的错误,如果你的集群安装的没有错误的话 ssh免登陆都可以的话(你的肯定是不行的) 请关闭你的集群节点的防火墙,并且正确安装ssh免登陆 -
hadoop2.x如何通过eclipse执行mapreduce[2023-04-11]
你参考下这个吧eclipse中开发Hadoop2.x的Map/Reduce项目汇总 -
用于hadoop mapreduce的罐子(Jars for hadoop mapreduce)[2022-10-19]
本教程提到: 下载Hadoop-core-1.2.1.jar,用于编译和执行MapReduce程序。 访问以下链接http://mvnrepository.com/artifact/org.apache.hadoop/hadoop-core/1.2.1下载jar。 所以在这里你可以找到不同版本的所有罐子 This tutorial mentions : Download Hadoop-core-1.2.1.jar, which is used to compile and execute the MapRe ... -
在Hadoop中,您处理输入拆分而不是块。 输入拆分是完整的数据集。 您希望避免一个映射器超过两个拆分的情况,因为这会降低性能并创建流量。 在文本世界中,假设您在block1中并且您有一个句子,例如“我是一个哈”,而block2继续“doop developer”,那么这会创建网络流量,因为我们始终必须在一个完整的节点上工作输入拆分和一些数据必须转移到另一个节点。 In Hadoop you work on input splits and not on blocks. An input split is ...
-
你的jar路径和你的Class名字在你的工作命令中是错误的 您写了/home/ciro/Scrivania/BDABI/wordcount/WordCountj.jar而不是/home/ciro/Scrivania/BDABI/wordcount/org/myorg/WordCountj.jar和wordcount而不是WordCount Your jar path and your Class name are wrong in your job command You wrote /home/ciro/ ...
-
好吧,它似乎是Cloudera Quickstart VM配置问题。 它适用于Hortonworks Sandbox。 不同之处在于: - hortonworks沙箱配置为使用“真实”IP来定位其节点 - cloudera VM配置为使用“localhost”IP来定位其节点(此处为数据节点)。 因此,当我尝试从Windows中的Cloudera虚拟集群中执行MR时,由资源管理器返回的配置是127.0.0.1而不是真实的IP,当然,我的本地计算机上没有任何数据节点。 Well, it seems to be ...
-
Hadoop和MapReduce(Hadoop and MapReduce)[2023-06-28]
Mapper接口按以下顺序需要4个类型参数:Map输入键,Map输入值,Map输出键和Map输出值。 在您的情况下,由于您正在处理4个整数,其中3个构成您的值,1个是您的密钥,因此使用IntWritable作为Map输入键并且应该使用Text而错误。 此外,您在MapClass定义中指定的类型与传递给Map函数的类型不匹配。 鉴于您正在处理文本文件,您的MapClass应定义如下: public static class MapClass extends MapReduceBase implements M ...