MapReduce概念
MapReduce是一种分布式计算模型，由谷歌提出，主要用于搜索领域，解决海量数据计算问题。 MR由两个阶段组成：Map和Reduce，用户只需要实现map()和reduce()两个函数实现分布式计算。 这两个函数的形参是key，value对，表示函数的输入信息。 MP执行流程 客户端提交给jobtracker，jobtracker分配给tasktracker。 trasktracker会对任务进行mapper和reducer操作。 MapReduce原理 一个map输入k1、v1，数据由输入文件中获取 map会把数据提交到每一个shuffle，最后输出到reducer任务。 reducer任务的数量跟mapper发送到shuffle的数量是一致的。 map任务处理 1.1、读取输入文件内容，解析成key，value对。对输入文件的每一个解析成key\value对。每个键值对调用一次map函数。 1.2、写自己的逻辑，对输入的key、value处理，转换成新的key、value输出。 1.3、对输出的key、value进行分区 1.4、对不同分区的数据按照key进行排序、分组。相同key的value放到一个集合中。 1.5、（可选）分组后对数据进行规约 reduce任务处理 2.1、对多个map任务的输出按照不同的分区通过网络拷贝到reduce节点。 2.2、对多个map任务输出进行合并、排序。写reduce函数自己的逻辑，对输入key、value处理，转换成新的key、value输出。 2.3、把输出的结果保存到HDFS中。 MapReduce执行过程 1.1.读取hdfs中文件，每个解析成&k,v&。每一个键值对调用一次map函数。 解析成两个&k,v&，分别&0,hello you&&10,hello me&。调用map函数两次。 k是每行的开始位置，v则示每行的文本内容。 1.2.覆盖map()函数，接收1.1产生的&k,v&，进行处理，转换新的&k,v&输出。 1.3.对1.2输出的&k,v&进行分区 public void map(k,v,context){ String[] split = v.toString().split(“ ”); for(String str : split){ context.write(str,1); } } 1.4.对不同分区中的数据进行排序(按照k)、分组，分别将key的value放到一个集合中。 map输出后的数据是：&hello,1]&,&you,1&,&hello,1&,&me,1& 排序后: &hello,1]&, &hello,1&,&you,1&,&me,1& 分组后：&hello,{1,1}&,&you,{1}&,&me,{1}& 1.5.（可选）对分组后的数据进行规约。 2.1.多个map任务的输出按照不同的分区，通过网络copy到不同的reduce节点上。 2.2.对多个map的输出进行合并，排序，覆盖reduce函数，接收的是分组的数据，实现自己的业务逻辑，处理后产生新的&k,v&输出。 reduce函数被调用3次，跟分组次数一致。 public void reduce(k,vs,context){ long sum =0L; for(long num:vs){ sum += } context.write(k,sum); } 2.3.设置任务执行，对reduce输出的&k,v&保存到hdfs中。 job.waitForCompletion(true);
整个流程我分了四步。简单些可以这样说，每个map task都有一个内存缓冲区，存储着map的输出结果，当缓冲区快满的时候需要将缓冲区的数据以一个临时文件的方式存放到磁盘，当整个map task结束后再对磁盘中这个map task产生的所有临时文件做合并，生成最终的正式输出文件，然后等待reduce task来拉数据。 MapReduce提交的源代码分析 waitForCompletion函数中的submit方法连接和提交到jobtracker。 在eclipse中写的代码如何提交到JobTracker中的哪？ 答（1）eclipse中调用的job.waitForCompletion(true),实际调用的是JobClient中的提交方法。 contect（） info = jobClient.submitJobInternal(conf) (2)在contect（）中，实际创建了一个JobClient对象，在调用该对象的构造方法时，获得了JobTracker的客户端代理对象JobSubmissionProtocol jobSubmissionProtocol实现类是JobTracker （3）在jobClient.submitJobInternal(conf)方法中，调用了jobSubmissionProtocol.submitJob() 即，执行的是JobTracker.submitJob(..) Hadoop基本类型 Hadoop数据类型必须实现Writable接口。 Long LongWritable Boolean BooleanWritable String Text Integer IntWritable Java类型转换为Hadoop基本类型： 直接调用hadoop类的构造方法，或者调用set（）方法 new IntWritable(123) Hadoop类型转换成Java类型： text需要调用toString方法 其他类型调用get()方法 使用Hadoop自定义类型处理手机上网流量 1、自定义类 class KpiWritable implements Writable{ long upPackN long downPackN long upPayL long downPayL public KpiWritable() {} public KpiWritable(String upPackNum,String downPackNum,String upPayLoad,String downPayLoad){ this.upPackNum = Long.parseLong(upPackNum); this.downPackNum = Long.parseLong(downPackNum); this.upPayLoad = Long.parseLong(upPayLoad); this.downPayLoad = Long.parseLong(downPayLoad); } @Override public void write(DataOutput out) throws IOException { //序列化出去 out.writeLong(upPackNum); out.writeLong(downPackNum); out.writeLong(upPayLoad); out.writeLong(downPayLoad); } @Override public void readFields(DataInput in) throws IOException { //顺序和写出去一样 this.upPackNum = in.readLong(); this.downPackNum = in.readLong(); this.upPayLoad = in.readLong(); this.downPayLoad = in.readLong(); } @Override public String toString() { return upPackNum+&\t&+downPackNum+&\t&+upPayLoad+&\t&+downPayL } 2、自定义Map static class MyMapper extends Mapper&LongWritable, Text, Text, KpiWritable&{ protected void map(LongWritable k1, Text v1, org.apache.hadoop.mapreduce.Mapper&LongWritable,Text,Text,KpiWritable&.Context context) throws IOException, InterruptedException { //处理接收的数据 String[] splits = v1.toString().split(&\t&); //获取手机号 String msisdn = splits[1]; Text k2 = new Text(msisdn); KpiWritable v2 = new KpiWritable(splits[6],splits[7],splits[8],splits[9]); //写入context中交过reduce执行 context.write(k2, v2); } } 3、自定义Reduce static class MyReduce extends Reducer&Text, KpiWritable, Text, KpiWritable&{ protected void reduce(Text k2, Iterable&KpiWritable& v2s,org.apache.hadoop.mapreduce.Reducer&Text, KpiWritable, Text, KpiWritable&.Context context) throws IOException, InterruptedException { /** * k2 表示不同的手机号 * v2s 表示该手机号不同时段流量集合 */ //定义计数器 long upPackNum = 0L; long downPackNum = 0L; long upPayLoad = 0L; long downPayLoad = 0L; //遍历合并数据 for(KpiWritable kpi : v2s){ upPackNum += kpi.upPackN downPackNum += kpi.downPackN upPayLoad += kpi.upPayL downPayLoad += kpi.downPayL } //封装到对象中 KpiWritable v3 = new KpiWritable(upPackNum+&&, downPackNum+&&, upPayLoad+&&, downPayLoad+&&); //写入context中 context.write(k2, v3); } } 4、写驱动程序 static final String INPUT_PATH = &hdfs://h1:9000/wlan&; static final String OUT_PATH = &hdfs://h1:9000/wlan_out&; public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); FileSystem fileSystem = FileSystem.get(new URI(&hdfs://h1:9000/&), conf); Path outPut = new Path(OUT_PATH); if(fileSystem.exists(outPut)){ fileSystem.delete(outPut,true); } /** * 1.1、指定输入文件路径 * 1.1.1. 指定那个类来格式化输入文 * 1.2、指定自定义的Mapper类 * 1.2.1.指定输出&k2,v2&的类型 * 1.3、指定分区 * 1.4、排序分区（TODO) * 1.5、（可选）合并 * 2.1、多个map任务的输出，通过网络copy到不同的reduce节点上，这个操作由hadoop自动完成 * 2.2、指定定义的reduce类 * 2.2.1.指定输出&k3,v3&类型 * 2.3、指定输出位置 * 2.3.1、设置输出文件的格式化类 * * 最后吧代码提交到JobTracker执行 */ //创建Job任务 Job job = new Job(conf,KpiApp.class.getSimpleName()); //设置输入路径 FileInputFormat.setInputPaths(job, INPUT_PATH); //设置输入数据使用的格式化类 job.setInputFormatClass(TextInputFormat.class); //设置自定义Map类 job.setMapperClass(MyMapper.class); //设置Map类输出的key和value值类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(KpiWritable.class); //设置分区类 job.setPartitionerClass(HashPartitioner.class); //设置任务数 job.setNumReduceTasks(1); //设置自定义Reduce类 job.setReducerClass(MyReduce.class); //设置输入数据使用的格式化类 job.setInputFormatClass(TextInputFormat.class); //设置Reduce输出的key和value值类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(KpiWritable.class); //设置输出文件位置 FileOutputFormat.setOutputPath(job, outPut); //设置将任务提交到JobTracker job.waitForCompletion(true); } MapReduce 0.x API区别 hadoop版本0.x 1、包一般是mapred。 2、使用是JobConf类创建Job任务。 3、使用JobClient.runJob(jobConf)提交任务。 4、自定义类需要继承MapReduceBase实现Mapper和Reducer接口。 hadoop版本1.x 1、包一般是mapreduce。 2、使用的Job类创建任务。 3、ob.waitForCompletion(true)提交任务。 4、自定义类只需要继承Mapper和Reducer类。 命令行运行指定参数 hadoop jar WordCount.jar hdfs://h1:9000/hello hdfs://h1:9000/cmd_out 跟eclipse直接运行的代码区别： 1、类需要继承org.apache.hadoop.conf.Configured，并实现org.apache.hadoop.util.Tool。 2、以前main方法中写的驱动程序卸载覆写的run方法中。3、run（）方法中 job.setJarByClass(CmdWordCount.class); 4、输入输出字符串定义为全局空字符串 5、main方法中使用org.apache.hadoop.util.ToolRunner的run方法，传入new CmdWordCount()和args。args是main方法接收的字符串数组。 6、在覆写的run方法中把接收到的args数组提取并赋值给INPUT和OUTPUT的路径 INPUT_PATH = agrs0[0]; OUTPUT_PATH = args0[1]; 6、打包时一定要记得选择输出类。
Hadoop计数器 File Input Format Counters Bytes Read=19 读取文件字节数 Map-Reduce Framework Map output materialized bytes=65 1#Map input records=2 读取记录行 Reduce shuffle bytes=65 Spilled Records=8 Map output bytes=51 Total committed heap usage (bytes)= 5#Combine input records=0 Map合并/规约输入 SPLIT_RAW_BYTES=85 3# Reduce input records=4 reduce输入行 4#Reduce input groups=3 Reduce输入组数 Combine output records=0 Map合并/规约输出 Reduce output records=3 Reduce输出记录 2#Map output records=4 Map输出行 通过计数器数可以检查出Map还是Reduce出现问题。 舆情监督示例，使用自定义计数器监控出现次数。 //自定义计数器 Counter helloCounter = context.getCounter(&Sensitive Words&, &hello&); String line = v1.toString(); if(line.contains(&hello&)){ helloCounter.increment(1L); }
Combine操作 为什么使用Combine？ Combiner发生在Map端。对数据进行规约处理，数据量变小了，传送到reduce的数据量变小，传输时间变短，作业整体时间变短。 为什么Combine不作为MapReduce的标配，而是可选配置？ 因为不是所有的算法都适合使用Combine处理，例如求平均数。 适用于求和 Combine本身已经执行了Reduce操作，为什么Reduce阶段还要执行reduce操作？ combine操作发送在map端，处理一个任务所接收的文件中的数据，不能跨map任务；只有reduce可以接收多个map任务处理数据。 设置参数 job.setCombinerClass(MyCombiner.class); MyCombiner.class可以使用Reduce
Partitioner编程 对输出的key，value进行分区。 指定自定义partition类，自定义类需要继承HashPartitioner类。覆盖getPartition方法。 分区的实例必须打成Jar包。 作用： 1、根据业务需要，产生多个输出文件。 2、多个reduce任务在运行，提高整体job的运行效率。 根据实际情况来使用，如果有5台机器，而分成100个分区来运行或出现延迟和整体效率低问题。因为需要排队运行！ 主要代码： 设置成打包运行 job.setJarByClass(KpiApp.class); 设置分区job.setPartitionerClass(MyPartitioner.class); 设置Reduce任务数 job.setNumReduceTask(2); 自定义Partitioner类需要继承HashPartition类，泛型使用K2，V2的类型。覆写getPartition方法。 排序和分组
在map和reduce阶段进行排序时，比较的是k2。v2是不参与排序比较。如果让v2也进行排序，需要将k2和v2组装成心的类，作为k2，才能参与比较。 新类需要实现WritableCompareble，覆写readFilds、write、compareTo、hasCode、equals方法。 在自定义map程序中将k2，v2封装到新类中，当做k2写入context。 编写驱动main方法时，设置map输出的类型（job.setMapOutputKeyClass(new.class)）
按照K2进行比较，这个k2是分装到自定义类的k2。 自定义分组类实现RewComparetor，覆写compare方法。 public int compare(NewK2 o1, NewK2 o2) { return (int) (o1.k2 - o2.k2); } /** * @param b1 表示第一个参与比较的字节数组 * @param s1 表示第一个参与比较的字节数组的起始位置 * @param l1 表示第一个参与比较的字节数组的偏移量 * * @param b2 表示第二个参与比较的字节数组 * @param s2 表示第二个参与比较的字节数组的起始位置 * @param l2 表示第二个参与比较的字节数组的偏移量 * */ public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { return WritableComparator.compareBytes(b1, s1, 8, b2, s2, 8); } job任务设置分组比较组 job.setGroupComparatorClass(MyGroup.class); Shuffle
MapReduce的核心，俗称洗牌、打乱。 shuffle在map任务传送到reduce任务之间。 Map端 1、每个map有一个环形内存缓冲区，用于存储任务的输出，默认100MB，如果达到法制80MB后台线程会把内容写到指定磁盘（mapred.local.dir）下的新建的溢出文件。 2、写入磁盘钱，要partition、sort。如果有combiner，combine后写入。 3、最后记录完成，合并全部溢写文件为一个分区且排序的文件。 Reduce端 1、Reduce通过Http方式得到输出文件的分区。 2、TaskTracker为分区文件运行Reduce任务。复制阶段把Map输出复制到Reducer的内存或磁盘。一个Map任务完成，Reduce开始复制输出。 3、排序阶段合并map输出，最后运行Reduce阶段。 MapReduce常见算法 单词计数 数据去重 排序 TopK 选择 投影 分组 多表连接 单边关联
阅读(...) 评论()> Hadoop中自定义计数器
Hadoop中自定义计数器
相关推荐：最近写了个mapReduce进行数据分析，发现数据不合理，想写一个计数器对某些发生异常的记录进行计数，但是查看了《hadoop权威指南》后发现使用的API是1点几 ，在网上搜到的也是1点几的例子，都用到了reporter，但是现在2.0后已经不用他，map 和reduce也是从mapR
I'm using elFinder Webfroms version from . I've downloaded and tried to run the project but unable to run the project. I'm getting following error while running the test project.
Server Error in '/' Application.
Configuration Error
Description: An error occurred during the processing of a configuration file required to service this request. Please review the specific error details below and modify your configuration file appropriately.
Parser Error Message: Required attribute 'duplicateFilePattern' not found.
Source Error:
Line 12: Line 13: Line 14: defaultVolumeName=&LocalFileSystem& baseUrl=&http://localhost:56439/data/& baseThumbsUrl=&http://localhost:56439/data/.thumbs/&& Line 16:
Source File: C:\Users\xyz\Desktop\elFinderFileManager\elFinderFileManager\web.config Line: 14
Version Information: Microsoft .NET Framework Version:4.0.30319; ASP.NET Version:4.0.
Error: 相关推荐：一、环境1、hadoop 0.20.22、操作系统Linux二、背景1、最近写MR的代码，总在想统计一些错误的数据出现的次数，发现如果都写在reduce的输出里太难看了，所以想找办法专门输出一些统计数字。2、翻看《hadoop权威指南》第8章第1节的时候发现能够自定义计数器，但
I've tried a lot to find on Google to find out the reason for this error but was not able to find any single clue on the web.
Please anyone know about it, let me know how can I resolve this issue?
Thanks in advance!
asked Oct 13 '15 at 17:44
2,764 6 21 35
1 Answer 1
Not 100% sure, as I can't find the docs easily, but the error is saying that you must specify a value for the duplicateFilePattern property.
A quick search on SO finds a common value for this is
duplicateFilePattern=&Copy of {0}&
suggesting it's a format string used to alter the name of a file in case it already exists.
I'd suggest checking your documentation for elFinder, specifically the section on configurations.
answered Oct 13 '15 at 17:54
100k 41 248 346 && & you're genius!, thanks. :) I've even checked and get idea from
this thread. –&
Oct 13 '15 at 18:03
Your Answer & draft saved draft discarded
Sign up or
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest Name Email
Post as a guest Name Email
By posting your answer, you agree to the
Not the answer you're looking for? Browse other questions tagged
From:http://blog.csdn.net/dajuezhao/article/details/5788705??一、环境1、hadoop 0.20.22、操作系统Linux二、背景1、最近写MR的代码，总在想统计一些错误的数据出现的次数...
------分隔线----------------------------
相关阅读排行
相关最新文章
Copyright 2012- ( Coin163 ) All Rights Reserved &&MapReduce 计数器简介
第2页_服务器应用_Linux公社-Linux系统门户网站
你好，游客
MapReduce 计数器简介
来源：Linux社区&
作者：Chuck_lee
Notice2： 使用计数器需要清楚的是它们都存储在jobTracker的内存里。 Mapper/Reducer 任务序列化它们，连同更新状态被发送。为了运行正常且jobTracker不会出问题，计数器的数量应该在10-100个，计数器不仅仅只用来聚合MapReduce job的统计值。新版本的限制了计数器的数量，以防给jobTracker带来损害。你最不想看到的事情就是由于定义上百个计数器而使jobTracker宕机。 &下面咱们来看一个计数器的实例（以下代码请运行在 0.20.1 版本以上）： &
3.1 测试数据：&hello world 2013 mapreducehello world 2013 mapreducehello world 2013 mapreduce 3.2 代码：&/**&* Project Name:CDHJobs&* File Name:MapredCounter.java&* Package Name:tmp&* Date:下午2:12:48&* Copyright (c) 2014,
All Rights Reserved.&*&*/
import java.io.IOEimport java.util.StringT
import mons.lang3.StringUimport org.apache.hadoop.conf.Cimport org.apache.hadoop.fs.Pimport org.apache.hadoop.io.IntWimport org.apache.hadoop.io.Timport org.apache.hadoop.mapreduce.Cimport org.apache.hadoop.mapreduce.CounterGimport org.apache.hadoop.mapreduce.Cimport org.apache.hadoop.mapreduce.Jimport org.apache.hadoop.mapreduce.Mimport org.apache.hadoop.mapreduce.Rimport org.apache.hadoop.mapreduce.lib.input.FileInputFimport org.apache.hadoop.mapreduce.lib.output.FileOutputF
public class WordCountWithCounter {
& static enum WordsNature {& & STARTS_WITH_DIGIT, STARTS_WITH_LETTER, ALL& }
* The map class of WordCount.&
*/& public static class TokenCounterMapper extends Mapper&Object, Text, Text, IntWritable& {
& & private final static IntWritable one = new IntWritable(1);& & private Text word = new Text();
& & public void map(Object key, Text value, Context context) throws IOException, InterruptedException {& & & StringTokenizer itr = new StringTokenizer(value.toString());& & & while (itr.hasMoreTokens()) {& & & & word.set(itr.nextToken());& & & & context.write(word, one);& & & }& & }& }
* The reducer class of WordCount&
*/& public static class TokenCounterReducer extends Reducer&Text, IntWritable, Text, IntWritable& {& & public void reduce(Text key, Iterable&IntWritable& values, Context context) throws IOException,& & & & InterruptedException {& & & int sum = 0;
& & & String token = key.toString();& & & if (StringUtils.isNumeric(token)) {& & & & context.getCounter(WordsNature.STARTS_WITH_DIGIT).increment(1);& & & } else if (StringUtils.isAlpha(token)) {& & & & context.getCounter(WordsNature.STARTS_WITH_LETTER).increment(1);& & & }& & & context.getCounter(WordsNature.ALL).increment(1);
& & & for (IntWritable value : values) {& & & & sum += value.get();& & & }& & & context.write(key, new IntWritable(sum));& & }& }
* The main entry point.&
*/& public static void main(String[] args) throws Exception {& & Configuration conf = new Configuration();& & Job job = new Job(conf, "WordCountWithCounter");& & job.setJarByClass(WordCountWithCounter.class);& & job.setMapperClass(TokenCounterMapper.class);& & job.setReducerClass(TokenCounterReducer.class);& & job.setOutputKeyClass(Text.class);& & job.setOutputValueClass(IntWritable.class);& & FileInputFormat.addInputPath(job, new Path("/tmp/dsap/rawdata/june/a.txt"));& & FileOutputFormat.setOutputPath(job, new Path("/tmp/dsap/rawdata/june/a_result"));& & int exitCode = job.waitForCompletion(true) ? 0 : 1;
& & Counters counters = job.getCounters();& & Counter c1 = counters.findCounter(WordsNature.STARTS_WITH_DIGIT);& & System.out.println("--------------&&&&: " + c1.getDisplayName() + ": " + c1.getValue());
& & // The below example shows how to get built-in counter groups that Hadoop provides basically.& & for (CounterGroup group : counters) {& & & System.out.println("==========================================================");& & & System.out.println("* Counter Group: " + group.getDisplayName() + " (" + group.getName() + ")");& & & System.out.println("& number of counters in this group: " + group.size());& & & for (Counter counter : group) {& & & & System.out.println("& ++++ " + counter.getDisplayName() + ": " + counter.getName() + ": "& & & & & & + counter.getValue());& & & }& & }& & System.exit(exitCode);& }} 3.3 结果与 计数器详解&运行结果下面会一并给出。Counter有"组group"的概念，用于表示逻辑上相同范围的所有数值。MapReduce job提供的默认Counter分为7个组，下面逐一介绍。这里也拿上面的测试数据来做详细比对，我将会针对具体的计数器，挑选一些主要的简述一下。 &... 前面省略 job 运行信息 xx 字 ...& ALL=4& STARTS_WITH_DIGIT=1& STARTS_WITH_LETTER=3--------------&&&&: STARTS_WITH_DIGIT: 1==========================================================#MapReduce job执行所依赖的数据来自于不同的文件系统，这个group表示job与文件系统交互的读写统计 * Counter Group: File System Counters (org.apache.hadoop.mapreduce.FileSystemCounter)&number of counters in this group: 10&#job读取本地文件系统的文件字节数。假定我们当前map的输入数据都来自于HDFS，那么在map阶段，这个数据应该是0。但reduce在执行前，它 的输入数据是经过shuffle的merge后存储在reduce端本地磁盘中，所以这个数据就是所有reduce的总输入字节数。&++++ FILE: Number of bytes read: FILE_BYTES_READ: 159&#map的中间结果都会spill到本地磁盘中，在map执行完后，形成最终的spill文件。所以map端这里的数据就表示map task往本地磁盘中总共写了多少字节。与map端相对应的是，reduce端在shuffle时，会不断地拉取map端的中间结果，然后做merge并 不断spill到自己的本地磁盘中。最终形成一个单独文件，这个文件就是reduce的输入文件。 &++++ FILE: Number of bytes written: FILE_BYTES_WRITTEN: 159447&++++ FILE: Number of read operations: FILE_READ_OPS: 0&++++ FILE: Number of large read operations: FILE_LARGE_READ_OPS: 0&++++ FILE: Number of write operations: FILE_WRITE_OPS: 0&# 整个job执行过程中，只有map端运行时，才从HDFS读取数据，这些数据不限于源文件内容，还包括所有map的split元数据。所以这个值应该比FileInputFormatCounters.BYTES_READ 要略大些。 &++++ HDFS: Number of bytes read: HDFS_BYTES_READ: 198&#Reduce的最终结果都会写入HDFS，就是一个job执行结果的总量。 &++++ HDFS: Number of bytes written: HDFS_BYTES_WRITTEN: 35&++++ HDFS: Number of read operations: HDFS_READ_OPS: 6&++++ HDFS: Number of large read operations: HDFS_LARGE_READ_OPS: 0&++++ HDFS: Number of write operations: HDFS_WRITE_OPS: 2==========================================================#这个group描述与job调度相关的统计 * Counter Group: Job Counters (org.apache.hadoop.mapreduce.JobCounter)&number of counters in this group: 5&#Job在被调度时，如果启动了一个data-local(源文件的幅本在执行map task的taskTracker本地) &++++ Data-local map tasks &#当前job为某些map task的执行保留了slot，总共保留的时间是多少 &++++ FALLOW_SLOTS_MILLIS_MAPS/REDUCES&#所有map task占用slot的总时间，包含执行时间和创建/销毁子JVM的时间&++++ SLOTS_MILLIS_MAPS/REDUCES&# 此job启动了多少个map task &++++ Launched map tasks: TOTAL_LAUNCHED_MAPS: 1&# 此job启动了多少个reduce task &++++ Launched reduce tasks: TOTAL_LAUNCHED_REDUCES: 1&++++ Rack-local map tasks: RACK_LOCAL_MAPS: 1&++++ Total time spent by all maps in occupied slots (ms): SLOTS_MILLIS_MAPS: 3896&++++ Total time spent by all reduces in occupied slots (ms): SLOTS_MILLIS_REDUCES: 9006==========================================================#这个Counter group包含了相当多地job执行细节数据。这里需要有个概念认识是：一般情况下，record就表示一行数据，而相对地byte表示这行数据的大小是 多少，这里的group表示经过reduce merge后像这样的输入形式{"aaa", [5, 8, 2, &]}。 * Counter Group: Map-Reduce Framework (org.apache.hadoop.mapreduce.TaskCounter)&number of counters in this group: 20&#所有map task从HDFS读取的文件总行数 &++++ Map input records: MAP_INPUT_RECORDS: 3&#map task的直接输出record是多少，就是在map方法中调用context.write的次数，也就是未经过Combine时的原生输出条数 &++++ Map output records: MAP_OUTPUT_RECORDS: 12&# Map的输出结果key/value都会被序列化到内存缓冲区中，所以这里的bytes指序列化后的最终字节之和 &++++ Map output bytes: MAP_OUTPUT_BYTES: 129&++++ Map output materialized bytes: MAP_OUTPUT_MATERIALIZED_BYTES: 159&# #与map task 的split相关的数据都会保存于HDFS中，而在保存时元数据也相应地存储着数据是以怎样的压缩方式放入的，它的具体类型是什么，这些额外的数据是 MapReduce框架加入的，与job无关，这里记录的大小就是表示额外信息的字节大小&++++ Input split bytes: SPLIT_RAW_BYTES: 117&#Combiner是为了减少尽量减少需要拉取和移动的数据，所以combine输入条数与map的输出条数是一致的。&++++ Combine input records: COMBINE_INPUT_RECORDS: 0&# 经过Combiner后，相同key的数据经过压缩，在map端自己解决了很多重复数据，表示最终在map端中间文件中的所有条目数 &++++ Combine output records: COMBINE_OUTPUT_RECORDS: 0&#Reduce总共读取了多少个这样的groups &++++ Reduce input groups: REDUCE_INPUT_GROUPS: 4&#Reduce端的copy线程总共从map端抓取了多少的中间数据，表示各个map task最终的中间文件总和 &++++ Reduce shuffle bytes: REDUCE_SHUFFLE_BYTES: 159&#如果有Combiner的话，那么这里的数值就等于map端Combiner运算后的最后条数，如果没有，那么就应该等于map的输出条数 &++++ Reduce input records: REDUCE_INPUT_RECORDS: 12&#所有reduce执行后输出的总条目数 &++++ Reduce output records: REDUCE_OUTPUT_RECORDS: 4&#spill过程在map和reduce端都会发生，这里统计在总共从内存往磁盘中spill了多少条数据 &++++ Spilled Records: SPILLED_RECORDS: 24&#每个reduce几乎都得从所有map端拉取数据，每个copy线程拉取成功一个map的数据，那么增1，所以它的总数基本等于 reduce number * map number &++++ Shuffled Maps : SHUFFLED_MAPS: 1&# copy线程在抓取map端中间数据时，如果因为网络连接异常或是IO异常，所引起的shuffle错误次数 &++++ Failed Shuffles: FAILED_SHUFFLE: 0&#记录着shuffle过程中总共经历了多少次merge动作 &++++ Merged Map outputs: MERGED_MAP_OUTPUTS: 1&#通过JMX获取到执行map与reduce的子JVM总共的GC时间消耗 &++++ GC time elapsed (ms): GC_TIME_MILLIS: 13&++++ CPU time spent (ms): CPU_MILLISECONDS: 3830&++++ Physical memory (bytes) snapshot: PHYSICAL_MEMORY_BYTES: &++++ Virtual memory (bytes) snapshot: VIRTUAL_MEMORY_BYTES: &++++ Total committed heap usage (bytes): COMMITTED_HEAP_BYTES: ==========================================================#这组内描述Shuffle过程中的各种错误情况发生次数，基本定位于Shuffle阶段copy线程抓取map端中间数据时的各种错误。* Counter Group: Shuffle Errors (Shuffle Errors)&number of counters in this group: 6&#每个map都有一个ID，如attempt__0254_m_，如果reduce的copy线程抓取过来的元数据中这个ID不是标准格式，那么此Counter增加 &++++ BAD_ID: BAD_ID: 0&#表示copy线程建立到map端的连接有误 &++++ CONNECTION: CONNECTION: 0&#Reduce的copy线程如果在抓取map端数据时出现IOException，那么这个值相应增加 &++++ IO_ERROR: IO_ERROR: 0&#map端的那个中间结果是有压缩好的有格式数据，所有它有两个length信息：源数据大小与压缩后数据大小。如果这两个length信息传输的有误(负值)，那么此Counter增加&++++ WRONG_LENGTH: WRONG_LENGTH: 0&#每个copy线程当然是有目的:为某个reduce抓取某些map的中间结果，如果当前抓取的map数据不是copy线程之前定义好的map，那么就表示把数据拉错了&++++ WRONG_MAP: WRONG_MAP: 0&#与上面描述一致，如果抓取的数据表示它不是为此reduce而准备的，那还是拉错数据了。 &++++ WRONG_REDUCE: WRONG_REDUCE: 0==========================================================#这个group表示map task读取文件内容(总输入数据)的统计 * Counter Group: File Input Format Counters (org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter)&number of counters in this group: 1# Map task的所有输入数据(字节)，等于各个map task的map方法传入的所有value值字节之和。 &++++ Bytes Read: BYTES_READ: 81==========================================================##这个group表示reduce task输出文件内容(总输出数据)的统计 * Counter Group: File Output Format Counters (org.apache.hadoop.mapreduce.lib.output.FileOutputFormatCounter)&number of counters in this group: 1&++++ Bytes Written: BYTES_WRITTEN: 35==========================================================# 自定义计数器的统计* Counter Group: tmp.WordCountWithCounter$WordsNature (tmp.WordCountWithCounter$WordsNature)&number of counters in this group: 3&++++ ALL: ALL: 4&++++ STARTS_WITH_DIGIT: STARTS_WITH_DIGIT: 1&++++ STARTS_WITH_LETTER: STARTS_WITH_LETTER: 3 4、最后的问题：&如果想要在 MapReduce 中实现一个类似计数器的&全局变量&，可以在 map、reduce 中以任意数据类型、任意修改变量值，并在 main 函数中回调获取该怎么办呢？
更多Hadoop相关信息见 专题页面
本文永久更新链接地址：2
相关资讯 & & &
& (02/27/:06)
& (02/24/:46)
& (11/29/:20)
& (05/26/:47)
& (12/24/:38)
　　　同意评论声明
　　　发表
尊重网上道德，遵守中华人民共和国的各项有关法律法规
承担一切因您的行为而直接或间接导致的民事或刑事法律责任
本站管理人员有权保留或删除其管辖留言中的任意内容
本站有权在网站内转载或引用您的评论
参与本评论即表明您已经阅读并接受上述条款
ltdh6889 发表于 Thank you very much ! It is very good and that help me solve a big problem! Thank you again!}