package org.jolirouge.ditp.ch02;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.output.FileOutputFormat;

// from figure 2.3, basic (primitive) word count

public class WordCount1 {
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = new Job(conf, "Primitive Word Counts");

		job.setJarByClass(WordCount1.class);

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

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

		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}

class WordCountMapper extends Mapper<IntWritable, Text, Text, IntWritable> {
	static final IntWritable one = new IntWritable(1);
	
	@Override
	public void map(IntWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		for (String token : value.toString().split(" ")) {
			context.write(new Text(token), one);
		}
	}
}

class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
	public void reduce(Text key, Iterable<IntWritable> values, Context context)
			throws IOException, InterruptedException {
		int count = 0;
		for (IntWritable value : values) {
			count += value.get();
		}
		context.write(key, new IntWritable(count));
	}
}

@@ -5,27 +5,8 @@
 age 
-org.jolirouge.ditp.
 ch02
@@ -157,16 +157,58 @@
 itable;%0A
+import org.apache.hadoop.io.LongWritable;%0A
 import o
@@ -1146,19 +1146,20 @@
  Mapper%3C
-Int
+Long
 Writable
@@ -1269,19 +1269,20 @@
 oid map(
-Int
+Long
 Writable

need better tokenization