package com.hadoop.Counter;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.conf.Configured;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import org.apache.hadoop.util.Tool;

import org.apache.hadoop.util.ToolRunner;

/**

 * @author Zimo

 *　ＭａｐＲｅｄｕｃｅ计数器

 */

public class MyCounter extends Configured implements Tool {

    /**

     * @param args

     */

    public static class MyCounterMap extends Mapper<LongWritable, Text, Text, Text> {

        //定义枚举对象

        public static enum LOG_PROCESSOR_COUNTER {

            //枚举对象BAD_RECORDS_LONG来统计长数据，枚举对象BAD_RECORDS_SHORT来统计短数据

            BAD_RECORDS_LONG, BAD_RECORDS_SHORT

        };

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

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

            if (arr_values.length > ) {

                //动态自定义计数器

                context.getCounter("ErrorCounter", "toolong").increment();

                //枚举声明计数器

                context.getCounter(LOG_PROCESSOR_COUNTER.BAD_RECORDS_LONG).increment();

            } else if(arr_values.length < ) {

                // 动态自定义计数器

                context.getCounter("ErrorCounter", "tooshort").increment();

                // 枚举声明计数器

                context.getCounter(LOG_PROCESSOR_COUNTER.BAD_RECORDS_SHORT).increment();

            } else {

                context.write(value, new Text(""));

            }

        }

    }

    public int run(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        Configuration conf = new Configuration();

        Path myPath = new Path(args[]);

        FileSystem hdfs = myPath.getFileSystem(conf);

        if (hdfs.isDirectory(myPath)) {

            hdfs.delete(myPath);

        }

        Job job = new Job(conf, "MyCounter");

        job.setJarByClass(MyCounter.class);

        job.setMapperClass(MyCounterMap.class);

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(args[]));

        FileOutputFormat.setOutputPath(job, new Path(args[]));

        job.waitForCompletion(true);

        return ;

    }

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

        // TODO Auto-generated method stub

        String[] arg0 = {

                "hdfs://centpy:9000/counter/counter.txt", "hdfs://centpy:9000/counter/out"

        };

        int ec = ToolRunner.run(new Configuration(), new MyCounter(), arg0);

        System.exit(ec);

    }

}

-- ::, INFO [org.apache.hadoop.mapreduce.Job] - Counters:

    File System Counters

        FILE: Number of bytes read=

        FILE: Number of bytes written=

        FILE: Number of read operations=

        FILE: Number of large read operations=

        FILE: Number of write operations=

        HDFS: Number of bytes read=

        HDFS: Number of bytes written=

        HDFS: Number of read operations=

        HDFS: Number of large read operations=

        HDFS: Number of write operations=

    Map-Reduce Framework

        Map input records=

        Map output records=

        Map output bytes=

        Map output materialized bytes=

        Input split bytes=

        Combine input records=

        Combine output records=

        Reduce input groups=

        Reduce shuffle bytes=

        Reduce input records=

        Reduce output records=

        Spilled Records=

        Shuffled Maps =

        Failed Shuffles=

        Merged Map outputs=

        GC time elapsed (ms)=

        CPU time spent (ms)=

        Physical memory (bytes) snapshot=

        Virtual memory (bytes) snapshot=

        Total committed heap usage (bytes)=

    ErrorCounter

        tooshort=

    com.hadoop.Counter.MyCounter$MyCounterMap$LOG_PROCESSOR_COUNTER

        BAD_RECORDS_SHORT=

    File Input Format Counters

        Bytes Read=

    File Output Format Counters

        Bytes Written=0