Hadoop中的计数器（累加器）

1、什么是计数器？

计数器是收集作业统计信息的有效手段之一，用于质量控制或应用级统计。计数器还可辅助诊断系统故障。如果需要将日志信息传输到map 或reduce 任务， 更好的方法通常是看能否用一个计数器值来记录某一特定事件的发生。对于大型分布式作业而言，使用计数器更为方便。除了因为获取计数器值比输出日志更方便，还有根据计数器值统计特定事件的发生次数要比分析一堆日志文件容易得多。

2、Hadoop内置计数器列表

每次mapreduce执行完成之后，我们都会看到一些日志记录出来，其中最重要的一些日志记录如下截图：

• 所有的这些都是MapReduce的计数器的功能，既然MapReduce当中有计数器的功能，我们如何实现自己的计数器？？？

3、需求：以前面文章介绍的排序以及序列化为案例，统计map接收到的数据记录条数。

4、实现：

第一种方式定义计数器：

• 通过context上下文对象可以获取我们的计数器，进行记录

• 通过context上下文对象，在map端使用计数器进行统计

  

   

    

     

    
    

     

      import org.apache.hadoop.io.LongWritable;
     
    
   

    

     

    
    

     

      import org.apache.hadoop.io.NullWritable;
     
    
   

    

     

    
    

     

      import org.apache.hadoop.io.Text;
     
    
   

    

     

    
    

     

      import org.apache.hadoop.mapreduce.Counter;
     
    
   

    

     

    
    

     

      import org.apache.hadoop.mapreduce.Mapper;
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      import java.io.IOException;
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      public 
      class FlowSortMapper extends Mapper<LongWritable, Text, FlowSortBean, NullWritable> {
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
    
      private FlowSortBean flowSortBean;
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
    
      //初始化
     
    
   

    

     

    
    

     
    
      @Override
     
    
   

    

     

    
    

     
    
      protected void setup(Context context) throws IOException, InterruptedException {
     
    
   

    

     

    
    

     

              flowSortBean = 
      new FlowSortBean();
     
    
   

    

     

    
    

     

          }
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
    
      @Override
     
    
   

    

     

    
    

     
    
      protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
     
    
   

    

     

    
    

     
        
      //自定义我们的计数器，这里实现了统计map输入数据的条数
     
    
   

    

     

    
    

     

              Counter counter = context.getCounter(
      "MR_COUNT", 
      "MapRecordCounter");
     
    
   

    

     

    
    

     

              counter.increment(
      1L);
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
        
      /**
     
    
   

    

     

    
    

     

       * 手机号 上行包 下行包 上行总流量 下行总流量
     
    
   

    

     

    
    

     

       * 13480253104 3 3 180 180
     
    
   

    

     

    
    

     

       */
     
    
   

    

     

    
    

     

              String[] split = value.toString().split(
      "\t");
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

              flowSortBean.setPhone(split[
      0]);
     
    
   

    

     

    
    

     

              flowSortBean.setUpPackNum(Integer.parseInt(split[
      1]));
     
    
   

    

     

    
    

     

              flowSortBean.setDownPackNum(Integer.parseInt(split[
      2]));
     
    
   

    

     

    
    

     

              flowSortBean.setUpPayLoad(Integer.parseInt(split[
      3]));
     
    
   

    

     

    
    

     

              flowSortBean.setDownPayLoad(Integer.parseInt(split[
      4]));
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

              context.write(flowSortBean, NullWritable.get());
     
    
   

    

     

    
    

     

          }
     
    
   

    

     

    
    

     

      }
     
    
  

运行程序之后就可以看到我们自定义的计数器在map阶段读取了七条数据 ：

第二种方式定义计数器：

通过enum枚举类型来定义计数器，

统计reduce端数据的输入的key有多少个，对应的value有多少个

  

   

    

     

    
    

     

      import org.apache.hadoop.io.NullWritable;
     
    
   

    

     

    
    

     

      import org.apache.hadoop.mapreduce.Reducer;
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      import java.io.IOException;
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      public 
      class FlowSortReducer extends Reducer<FlowSortBean, NullWritable, FlowSortBean, NullWritable> {
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
    
      public 
      static 
      enum Counter{
     
    
   

    

     

    
    

     

              REDUCE_INPUT_RECORDS
     
    
   

    

     

    
    

     

          }
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     
    
      @Override
     
    
   

    

     

    
    

     
    
      protected void reduce(FlowSortBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
     
    
   

    

     

    
    

     

              context.getCounter(Counter.REDUCE_INPUT_RECORDS).increment(
      1L);
     
    
   

    

     

    
    

     
        
      //经过排序后的数据，直接输出即可
     
    
   

    

     

    
    

     

              context.write(key, NullWritable.get());
     
    
   

    

     

    
    

     

          }
     
    
   

    

     

    
    

     

      }