/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package mapreduce;

import java.io.IOException;
import java.util.*;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.mapreduce.*;
//import org.apache.hadoop.util.*;
import org.apache.log4j.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;

/**
 *
 * @author sc-md
 */
public class MapReduceWordCount
{
    public static class TokenizerMapper
       extends Mapper<Object, Text, Text, IntWritable>{

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

    @Override
    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
        String line = value.toString();
      //StringTokenizer itr = new StringTokenizer(line);
      String[] result = line.split("\\s");

      Logger.getLogger("TokenizerMapper").debug("Read string " + line );
//      while (itr.hasMoreTokens()) {
//        word.set(itr.nextToken());
//        context.write(word, one);
//      }
      for (int x = 0; x < result.length; x++)
      {
        word.set(result[x]);
        context.write(word, one);         
      }
    }
  }
    
  public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();

    @Override
    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }
  
    public static void Execute(Configuration conf, Job job)throws Exception
    {               
        //job.setMapOutputKeyClass(Text.class);               // Specifically set outputs for Mapper class..
        //job.setMapOutputValueClass(IntWritable.class);      // ... alternatively use setOutputKey/Value to set both Mapper and Reducer to same types
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        job.setMapperClass(MapReduceWordCount.TokenizerMapper.class);   
        job.setCombinerClass(MapReduceWordCount.IntSumReducer.class);
        job.setReducerClass(MapReduceWordCount.IntSumReducer.class);
        
//        job.setInputFormatClass(TextInputFormat.class);
//        job.setOutputFormatClass(TextOutputFormat.class);

        //job.setUser("hadoop");
        FileInputFormat.addInputPath(job, new Path("/mapreducedemo/textfiles"));
        FileOutputFormat.setOutputPath(job, new Path("/mapreducedemo/results5"));

        // Erase previous results (using HDFS)
        FileSystem.get(conf).delete(new Path("/mapreducedemo/results5"), true);
    }
    
    public static void OnCompletion(Configuration conf) throws Exception
    {
        // Do Nothing
    }
    
}