package org.hackreduce.examples;

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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.hackreduce.mappers.ModelMapper;


/**
 * This MapReduce job will count the total number of Bixi records in the data dump.
 *
 */
public abstract class RecordCounter extends Configured implements Tool {

	public enum RecordCounterCount {
		UNIQUE_KEYS
	}

	public static class RecordCounterReducer extends Reducer<Text, LongWritable, Text, LongWritable> {

		@Override
		protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
			context.getCounter(RecordCounterCount.UNIQUE_KEYS).increment(1);

			long count = 0;
			for (LongWritable value : values) {
				count += value.get();
			}

			context.write(key, new LongWritable(count));
		}

	}

	public abstract Class<? extends ModelMapper<?,?,?,?,?>> getMapper();

	public abstract void configureJob(Job job);

	@Override
	public int run(String[] args) throws Exception {
		Configuration conf = getConf();

        if (args.length != 2) {
        	System.err.println("Usage: " + getClass().getName() + " <input> <output>");
        	System.exit(2);
        }

        // Creating the MapReduce job (configuration) object
        Job job = new Job(conf);
        job.setJarByClass(getClass());
        job.setJobName(getClass().getName());

        // Tell the job which Mapper and Reducer to use (classes defined above)
        job.setMapperClass(getMapper());
		job.setReducerClass(RecordCounterReducer.class);

        configureJob(job);

		// This is what the Mapper will be outputting to the Reducer
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(LongWritable.class);

		// This is what the Reducer will be outputting
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);

		// Setting the input folder of the job 
		FileInputFormat.addInputPath(job, new Path(args[0]));

		// Preparing the output folder by first deleting it if it exists
        Path output = new Path(args[1]);
        FileSystem.get(conf).delete(output, true);
	    FileOutputFormat.setOutputPath(job, output);

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

@@ -645,13 +645,8 @@
  of 
-Bixi 
 reco
@@ -656,21 +656,21 @@
  in 
-the data dump
+an input file
 .%0A *
@@ -1977,36 +1977,8 @@
 );%0A%0A
-        configureJob(job);%0A%0A
 %09%09//
@@ -2257,24 +2257,95 @@
 le.class);%0A%0A
+%09%09// Do this last so subclass can override%0A        configureJob(job);%0A%0A
 %09%09// Setting

Move call to subclasses configureJob() method later so that it can override all defaults.