Hadoop 计数器

计数器

1）计数器主要用来收集系统信息，以及相关作业的运行时候的统计数据，用于知道作业成功、失败等情况。

2）相比而言，计数器方式比日志更易于分析

内置计数器

1）Hadoop内置的计数器，主要用来记录作业的执行情况

2）内置计数器包括：

A、MapReduce框架计数器（Map-Reduce Framework）

B、文件系统计数器（FileSystemCounters）

C、作业计数器（Job Counters）

D、文件输出格式计数器（File Output Format Counters）

E、文件输入格式计数器（File Input Format Counters）

3）计数器由相关的task进行维护，每个任务更新自己的计数器值，定期传递给TaskTracker，再由TaskTracker传给JobTracker

4）最终的作业计数器实际上是由JobTracker维护，所以计数器可以被全局汇总，同时也不必在整个网络中传递

5）只有当一个作业执行完成后，最终的计数器的值才是完整可靠的。

自定义Java计数器

1）MapReduce框架允许用户自定义计数器

2）计数器是一个全局变量

3）计数器有组的概念，可以用Java的枚举类型或者用字符串来定义

public Counter getCounter(Enum<?> counterName)
public Counter getCounter(String groupName,String counterName)

4）字符串方式（动态计数器）比枚举类型要更加灵活，可以动态在一个组下面添加多个计数器

5）计数器递增

public void increment(long incr)

计数器使用

1）Web UI查看

2）命令行方式： hadoop job -counter

3）使用Hadoop API

通过job.getCounters()得到Counters，而后调用counters.findCounter()方法去获得计数器对象。查看最终的计数器的值需要等作业完成后。

举个例子：

MyMapper.java

package com.counter;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MyMapper extends Mapper<LongWritable, Text, LongWritable, Text> {

@Override
protected void map(LongWritable key, Text value,Context context)
throws IOException, InterruptedException {
String[] val = value.toString().split("\\s+") ;
if(val.length < 2 ){
context.getCounter("ErrorCounter", "below_2").increment(1) ;
}else if(val.length > 2 ){
context.getCounter("ErrorCounter", "above_2").increment(1) ;
}
context.write(key, value) ;
}

}

TestCounter.java

package com.counter;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class TestCounter {
public static void main(String args[]) throws Exception{
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args)
.getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage: wordcount <in> <out>");
System.exit(2);
}
Job job = new Job(conf, "word count");
job.setJarByClass(TestCounter.class);
job.setMapperClass(MyMapper.class);
job.setNumReduceTasks(0) ;

//      job.setOutputKeyClass(Text.class);
//      job.setOutputValueClass(IntWritable.class);

FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
if(job.waitForCompletion(true) ){
Counter counter1 = job.getCounters().findCounter("ErrorCounter", "below_2") ;
System.out.println("长度小于2的记录数："+counter1.getValue()) ;
Counter counter2 = job.getCounters().findCounter("ErrorCounter", "above_2") ;
System.out.println("长度大于2的记录数："+counter2.getValue()) ;
System.out.println("作业运行成功");
System.exit(0) ;
}else{
System.out.println("作业运行失败");
System.exit(1) ;
}

}
}