9.1hadoop 内置计数器、自定义枚举计数器、Streaming计数器

标签：other   ota   EAP   cdc   状态   epo   extends   运行   static   

1.1  计数器

计数器的作用是用来统计数量的，用于记录特定事件的次数，分为内置计数器、自定义java枚举计数器、自定义Stream计数器三大类。用于质量分析，或应用级统计。分析计数器的值比分析一堆日志更高效。

1.1.1         内置计数器

（1）   任务计数器

MAP_INPUT_RECORDS计数器统计map任务输入记录的总数，一个作业的所有map任务进行聚集，定期发送给application master，全量发送更新。可以统计任务输入、输出的记录数量，输入、输出的字节数，分片（split）的原始字节数(SPLIT_RAW_BYTES)等

表内置的mapreduce任务计数器

（2）   作业计数器

作业计数器有application master维护，作业级别统计，值不会随着任务的执行而变化，例如TOTAL_LAUNCHED_MAPS统计作业执行的任务数。

1.1.2         自定义java计数器

计数器由java枚举类型来定义，以便进行分组，枚举名称即为组名，字段即为计数器，计数器为全局的，mapreduce框架跨所有map和reduce聚集这些计数器。

import java.io.IOException;

import java.util.Iterator;

import org.apache.hadoop.conf.Configured;

import org.apache.hadoop.io.*;

import org.apache.hadoop.mapred.JobClient;

import org.apache.hadoop.mapred.JobConf;

import org.apache.hadoop.mapred.MapReduceBase;

import org.apache.hadoop.mapred.Mapper;

import org.apache.hadoop.mapred.OutputCollector;

import org.apache.hadoop.mapred.Reducer;

import org.apache.hadoop.mapred.Reporter;

import org.apache.hadoop.util.Tool;

import org.apache.hadoop.util.ToolRunner;

//统计最高气温的作业。也统计气温值缺少的记录，不规范的记录

public class MaxTemperatureWithCounters extends Configured implements Tool {

 //自定义计数器

       enum Temperature {

              MiSSING, //统计气温缺失的记录

MALFORMED//统计不规则的记录

       }

 //定义maper

       static class MaxTemeratureMapperWithCounters extends MapReduceBase implements

                     Mapper<LongWritable, Text, Text, IntWritable> {

              private NcdcRecordParser parser = new NcdcRecordParser();

              @Override

              public void map(LongWritable key, Text value,

                            OutputCollector<Text, IntWritable> output, Reporter reporter)

                            throws IOException {

                     parser.parse(value);

                     if (parser.isValidTemperature()) {

                            int airTemperature = parser.getAirTemperature();

                            output.collect(new Text(parser.getYear()), new IntWritable(

                                          airTemperature));

                     } else if (parser.isMa1formedTemperature()) {

       //增加计数器的值                  //context.getCounter(Temperature.MALFORMED).increment(1);

                            // Reporter是MapReduce提供给应用程序的工具。可使用Reporter中的方法报告完成进度(progress)、设定状态消息(setStatus)以及更新计数器(incrCounter)

reporter.incrCounter(Temperature.MALFORMED, 1);

                     } else if (parser.IsMissingTemperature()) {

//context.getCounter(Temperature.MISSING).increment(1);

                            reporter.incrCounter(Temperature. MISSING, 1);

                     }

                    //动态计数器     context.getCounter(“TemperatureQuality”,parse.getQuality()).increment(1);

              }

       }

 //定义reduce

       static class MaxTemperatureReduceWithCounters extends MapReduceBase implements

                     Reducer<Text, IntWritable, Text, IntWritable> {

              public void reduce(Text key, Iterator<IntWritable> values,

                            OutputCollector<Text, IntWritable> output, Reporter reporter)

                            throws IOException {

                     int maxValue = Integer.MIN_VALUE;

                     while (values.hasNext()) {

                            maxValue = Math.max(maxValue, values.next().get());

                     }

                     output.collect(key, new IntWritable(maxValue));

              }

       }

       @Override

       public int run(String[] args) throws Exception {

              args = new String[] { "/test/input/t", "/test/output/t" }; // 给定输入输出路径

              JobConf conf = JobBuilder.parseInputAndOutput(this, getConf(), args);

              if (conf == null) {

                     return -1;

              }

              conf.setOutputKeyClass(Text.class);

              conf.setOutputValueClass(IntWritable.class);

              conf.setMapperClass(MaxTemeratureMapperWithCounters.class);

             conf.setCombinerClass(MaxTemperatureReduceWithCounters.class);

              conf.setReducerClass(MaxTemperatureReduceWithCounters.class);

              JobClient.runJob(conf);

              return 0;

       }

       public static void main(String[] args) throws Exception {

              int exitCode = ToolRunner.run(new MaxTemperatureWithCounters(), args);

              System.exit(exitCode);

       }

}

执行任务，-counters参数，会输出所有计数器的值

hadoop jar Hadoop-examples.jar MaxTemperatureWithCounter input/ncdc/all output -counters

（1）动态计数器   

动态计数器不像枚举型计数器需要提前定义组名和计数器类型，而是通过字符串名称动态的创建计数器。枚举类型计数器传入枚举类型也要转成String，所以两种方法时等价的，枚举型简单安全。

context.getCounter(String groupName,String counterName);

context.getCounter(“TemperatureQuality”,parse.getQuality()).increment(1);

（2）获取计数器

或者用javaAPI获取计数器值。根据配置信息创建cluster对象，根据jobid获取job，获取job的计数器，根据类路径获取计数器的值，counters.findCounter(MaxtemperatureCounters.Temperature.MISSING).getValue();方法。

public class MissingTemperatureFields extends Configured implement Tool{

@override

public int run（String[] args）throws Exception{

              if(args.length（）！=1)

{

       return -1；

}

String jobID=args[0];

//根据配置信息创建cluster对象

Cluster cluster =new Cluster(getConf());

//根据jobid获取job

 Job job=cluster.getJob(JobID.forName(jobID));

if(job==null)

{

System.err.printf(“NO job with ID %s”,jobID);

return -1;

}

if(!job.isComplete())

{

System.err.printf(“job ID %s is not complete”,jobID);

return -1;

}

获取job的计数器，

Counters counters=job.getCounters();

//根据路径获取计数器的值

long missing=counters.findCounter(MaxtemperatureCounters.Temperature.MISSING).getValue();

long total=counters.finCounter(TaskCounter.MAP_INPUT_RECORDS).getValue();

//计算任务的失败率

System.out.printf(“Records missing present %.2f/%%\n”,100.0*missing/total);

}

public static void main(String[] args)throw Exception{

       int exitCode=ToolRunner.run(new MissingTemperatureFields(),args);

       System.exit(exitCode);

}

}

执行hadoop任务，传入jobid参数

%hadoop jar hadoop-example.jar MissingTempratureFields job_11223131_0007

1.1.3         用户自定义的Streaming计数器

Hadoop streaming是hadoop的一个工具，用于运行非java的maper或reducer作业，例如maper和reducer是C++编写的可执行程序或者脚本文件。使用Streaming的mapreduce程序可以像保准错误流发送特殊格式的信息，增加计数器的值格式如下：reporter:counter:group.counter,amount

Python实例如下

sys.stderr.write(“reporter:counter:Temperature,Missing,1 \n”);

状态信息发送格式如下

sys.stderr.write(“reporter:status:message”);

1.1.4         获取计数器的方法汇总

1）web界面查看计数器值

2）命令行mapred job –counter查看计数器的值；

3）动态计数器用context获取，context.getCounter (“TemperatureQuality”,parse.getQuality()).increment(1);

4）用context的getCounter方法或者用reportor的incrCounter方法context.getCounter(Temperature.MISSING).increment(1);reporter.incrCounter(Temperature. MISSING, 1);

5）Streaming程序，即非java的mapreduce程序，通过向标准输出发送固定格式的数据来增加计数器的值。sys.stderr.write(“reporter:counter:Temperature,Missing,1 \n”);

6）或者用javaAPI获取计数器值。根据配置信息创建cluster对象，根据jobid获取job，获取job的计数器，根据类路径获取计数器的值，counters.findCounter(MaxtemperatureCounters.Temperature.MISSING).getValue();

自己开发了一个股票智能分析软件，功能很强大，需要的点击下面的链接获取：

https://www.cnblogs.com/bclshuai/p/11380657.html

9.1hadoop 内置计数器、自定义枚举计数器、Streaming计数器

标签：other   ota   EAP   cdc   状态   epo   extends   运行   static   

原文地址：https://www.cnblogs.com/bclshuai/p/12297802.html