content: 
  /**
   * MapReduce program for parsing XML-like data from an IRC chatroom
   * This program groups all messages by user. Designed to work with Hadoop 2.4.
   * 
   * @author Albert Haque
   * @date April 2014
   */
  
  import java.io.IOException;
  
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.conf.*;
  import org.apache.hadoop.io.*;
  import org.apache.hadoop.mapreduce.*;
  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;
  
  public class UserCounter {
  
  	public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
  		
  		private static final IntWritable ONE = new IntWritable(1);
  		
  		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  			String line = value.toString();
  			context.write(new Text(line.substring(line.indexOf("<user>")+6, line.indexOf("</user>"))), ONE);
  		}
  		
  	}
  
  	public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {
  
  		public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
  			int sum = 0;
  			for (IntWritable iw : values) {
  				sum += iw.get();
  			}
  			context.write(key, new IntWritable(sum));
  		}
  	}
  
  	public static void main(String[] args) throws Exception {
  		Configuration conf = new Configuration();
  		@SuppressWarnings("deprecation")
  		Job job = new Job(conf, "UserCounter");
  
  		job.setOutputKeyClass(Text.class);
  		job.setOutputValueClass(IntWritable.class);
  
  		job.setMapperClass(Map.class);
  		job.setReducerClass(Reduce.class);
  
  		job.setInputFormatClass(TextInputFormat.class);
  		job.setOutputFormatClass(TextOutputFormat.class);
  
  		FileInputFormat.addInputPath(job, new Path(args[0]));
  		FileOutputFormat.setOutputPath(job, new Path(args[1]));
  
  		job.waitForCompletion(true);
  	}
  
  }