Word Count 0.20 - Denis

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.ToolRunner;

import com.bejoy.sampels.worcount.WordCountDriver;

public class WordCountNewAPI {

    public static class WordCountMapper extends Mapper<LongWritable,Text, Text, IntWritable> {

            private final static IntWritable ONE = new IntWritable(1);
            private Text word = new Text();

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

                    String line = value.toString();
                    StringTokenizer tokenizer = new StringTokenizer(line);
                    while (tokenizer.hasMoreTokens()) {
                            word.set(tokenizer.nextToken());
                            context.write(word, ONE);
                    }

                    super.map(key, value, context);
            }

    }

    public static class WordCountReducer extends Reducer<Text,IntWritable, Text, IntWritable> {

            @Override
            protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

                    int sum = 0;
                    Iterator<IntWritable> iter = values.iterator();
                    while (iter.hasNext()) {
                            sum += iter.next().get();
                    }

                    context.write(key, new IntWritable(sum));

                    super.reduce(key, values, context);
            }

    }

    /**
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {


    	 int res = ToolRunner.run(new Configuration(), new WordCountDriver(), args);
         System.exit(res);
    }

    public Void run() throws Exception {

        Configuration conf = new Configuration();
        conf.set("mapred.job.tracker", "<ip:8021>");

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

        job.setJarByClass(WordCountNewAPI.class);

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

        job.setMapperClass(WordCountMapper.class);
        job.setCombinerClass(WordCountReducer.class);
        job.setReducerClass(WordCountReducer.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        FileInputFormat.setInputPaths(job, new Path("hdfs://localhost:9000/userdata/bejoy/samples/wc/input"));
        FileOutputFormat.setOutputPath(job, new Path("hdfs://localhost:9000/userdata/bejoy/samples/wc/input"));

                boolean b = job.waitForCompletion(true);
                if (!b) {
                        throw new IOException("error with job!");
                }

        return null;
    }

}