MapReduce 计数器简介
一、计数器简介
在许多状况下，一个用户须要了解待分析的数据，尽管这并不是所要执行的分析任务 的核心内容。以统计数据集中无效记录数目的任务为例，若是发现无效记录的比例 至关高，那么就须要认真思考为什么存在如此多无效记录。是所采用的检测程序存在 缺陷，仍是数据集质量确实很低，包含大量无效记录？若是肯定是数据集的质量问 题，则可能须要扩大数据集的规模，以增大有效记录的比例，从而进行有意义的 分析。计数器是一种收集做业统计信息的有效手段，用于质量控制或应用级统计。计数器 还可辅助诊断系统故障。若是须要将日志信息传输到map或reduce任务，更好的 方法一般是尝试传输计数器值以监测某一特定事件是否发生。对于大型分布式做业 而言，使用计数器更为方便。首先，获取计数器值比输出日志更方便，其次，根据 计数器值统计特定事件的发生次数要比分析一堆日志文件容易得多。
2、内置计数器
Hadoop为每一个做业维护若干内置计数器, 以描述该做业的各项指标。例如，某些计数器记录已处理的字节数和记录数，使用户可监控已处理的输入数据量和已产生的输出数据量，并以此对 job 作适当的优化。 html
14/06/08 15:13:35 INFO mapreduce.Job: Counters: 46 File System Counters FILE: Number of bytes read=159 FILE: Number of bytes written=159447 FILE: Number of read operations=0 FILE: Number of large read operations=0 FILE: Number of write operations=0 HDFS: Number of bytes read=198 HDFS: Number of bytes written=35 HDFS: Number of read operations=6 HDFS: Number of large read operations=0 HDFS: Number of write operations=2 Job Counters Launched map tasks=1 Launched reduce tasks=1 Rack-local map tasks=1 Total time spent by all maps in occupied slots (ms)=3896 Total time spent by all reduces in occupied slots (ms)=9006 Map-Reduce Framework Map input records=3 Map output records=12 Map output bytes=129 Map output materialized bytes=159 Input split bytes=117 Combine input records=0 Combine output records=0 Reduce input groups=4 Reduce shuffle bytes=159 Reduce input records=12 Reduce output records=4 Spilled Records=24 Shuffled Maps =1 Failed Shuffles=0 Merged Map outputs=1 GC time elapsed (ms)=13 CPU time spent (ms)=3830 Physical memory (bytes) snapshot=537718784 Virtual memory (bytes) snapshot=7365263360 Total committed heap usage (bytes)=2022309888 Shuffle Errors BAD_ID=0 CONNECTION=0 IO_ERROR=0 WRONG_LENGTH=0 WRONG_MAP=0 WRONG_REDUCE=0 File Input Format Counters Bytes Read=81 File Output Format Counters Bytes Written=35计数器由其关联任务维护，并按期传到tasktracker，再由tasktracker传给 jobtracker.所以，计数器可以被全局地汇集。详见第 hadoop 权威指南第170页的“进度和状态的更新”小节。与其余计数器（包括用户定义的计数器）不一样，内置的做业计数器实际上 由jobtracker维护，没必要在整个网络中发送。
一个任务的计数器值每次都是完整传输的，而非自上次传输以后再继续数未完成的传输，以免因为消息丢失而引起的错误。另外，若是一个任务在做业执行期间失 败，则相关计数器值会减少。仅当一个做业执行成功以后，计数器的值才是完整可 靠的。
三、用户定义的Java计数器
MapReduce容许用户编写程序来定义计数器，计数器的值可在mapper或reducer 中增长。多个计数器由一个Java枚举(enum)类型来定义，以便对计数器分组。一 个做业能够定义的枚举类型数量不限，各个枚举类型所包含的字段数量也不限。枚 举类型的名称即为组的名称，枚举类型的字段就是计数器名称。计数器是全局的。 换言之，MapReduce框架将跨全部map和reduce汇集这些计数器，并在做业结束 时产生一个最终结果。Notice1：须要说明的是，不一样的 hadoop 版本定义的方式会有些许差别。 java
（1）在0.20.x版本中使用counter很简单,直接定义便可，如无此counter，hadoop会自动添加此counter.
shell
Counter ct = context.getCounter("INPUT_WORDS", "count");
ct.increment(1);
（2）在0.19.x版本中,须要定义enum
enum MyCounter {INPUT_WORDS };
reporter.incrCounter(MyCounter.INPUT_WORDS, 1);
RunningJob job = JobClient.runJob(conf);
Counters c = job.getCounters();
long cnt = c.getCounter(MyCounter.INPUT_WORDS);
Notice2：使用计数器须要清楚的是它们都存储在jobTracker的内存里。Mapper/Reducer 任务序列化它们，连同更新状态被发送。为了运行正常且jobTracker不会出问题，计数器的数量应该在10-100个，计数器不只仅只用来聚合MapReduce job的统计值。新版本的hadoop限制了计数器的数量，以防给jobTracker带来损害。你最不想看到的事情就是因为定义上百个计数器而使jobTracker宕机。
下面我们来看一个计数器的实例（如下代码请运行在 0.20.1 版本以上）：
apache
3.1 测试数据：
hello world 2013 mapreduce hello world 2013 mapreduce hello world 2013 mapreduce
3.2 代码：
/**
* Project Name:CDHJobs
* File Name:MapredCounter.java
* Package Name:tmp
* Date:2014-6-8下午2:12:48
* Copyright (c) 2014, decli#qq.com All Rights Reserved.
*
*/
package tmp;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.CounterGroup;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
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: 537718784
++++ Virtual memory (bytes) snapshot: VIRTUAL_MEMORY_BYTES: 7365263360
++++ Total committed heap usage (bytes): COMMITTED_HEAP_BYTES: 2022309888
==========================================================
#这组内描述Shuffle过程当中的各类错误状况发生次数，基本定位于Shuffle阶段copy线程抓取map端中间数据时的各类错误。
* Counter Group: Shuffle Errors (Shuffle Errors)
number of counters in this group: 6
#每一个map都有一个ID，如attempt_201109020150_0254_m_000000_0，若是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
四、最后的问题：
若是想要在 MapReduce 中实现一个相似计数器的“全局变量”，能够在 map、reduce 中以任意数据类型、任意修改变量值，并在 main 函数中回调获取该怎么办呢？ app
五、Refer:
（1）An Example of Hadoop MapReduce Counter 框架
http://diveintodata.org/2011/03/15/an-example-of-hadoop-mapreduce-counter/ 分布式
（2）Hadoop Tutorial Series, Issue #3: Counters In Action ide
http://www.philippeadjiman.com/blog/2010/01/07/hadoop-tutorial-series-issue-3-counters-in-action/ 函数
（3）Controlling Hadoop MapReduce Job recursion
http://codingwiththomas.blogspot.com/2011/04/controlling-hadoop-job-recursion.html
（4）MapReduce Design Patterns（chapter 2 （part 3））（四）
http://blog.csdn.net/cuirong1986/article/details/8456923
（5）[hadoop源码阅读][5]-counter的使用和默认counter的含义
http://www.cnblogs.com/xuxm2007/archive/2012/06/15/2551030.html
相关文章
- 1. MapReduce 计数器简介
- 2. MapReduce计数器
- 3. MapReduce之计数器
- 4. MapReduce 简介
- 5. MapReduce简介
- 6. mapreduce简介
- 7. mapreduce实现简单计数
- 8. MapReduce 简单数据统计
- 9. MapReduce--2--MapReduce全局计数器
- 10. HDFS、YARN、Mapreduce简介
- 更多相关文章...
- • MySQL触发器简介 - MySQL教程
- • 浏览器 统计 - 浏览器信息
- • Github 简明教程
- • 使用Rxjava计算圆周率
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. No provider available from registry 127.0.0.1:2181 for service com.ddbuy.ser 解决方法
- 2. Qt5.7以上调用虚拟键盘(支持中文)，以及源码修改(可拖动，水平缩放)
- 3. 软件测试面试- 购物车功能测试用例设计
- 4. ElasticSearch（概念篇）：你知道的, 为了搜索…
- 5. redux理解
- 6. gitee创建第一个项目
- 7. 支持向量机之硬间隔（一步步推导，通俗易懂）
- 8. Mysql 异步复制延迟的原因及解决方案
- 9. 如何在运行SEPM配置向导时将不可认的复杂数据库密码改为简单密码
- 10. windows系统下tftp服务器使用
欢迎关注本站公众号,获取更多信息
相关文章
- 1. MapReduce 计数器简介
- 2. MapReduce计数器
- 3. MapReduce之计数器
- 4. MapReduce 简介
- 5. MapReduce简介
- 6. mapreduce简介
- 7. mapreduce实现简单计数
- 8. MapReduce 简单数据统计
- 9. MapReduce--2--MapReduce全局计数器
- 10. HDFS、YARN、Mapreduce简介