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 MessageGrouper {
  
  	public static class Map extends Mapper<LongWritable, Text, Text, Text> {
  		
  		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  			String line = value.toString();
  			// Extract the username
  			int usernameStartIndex = line.indexOf("<user>")+6;
  			int usernameEndIndex = line.indexOf("</user>");
  			// Key is username, message is value
  			context.write(new Text(line.substring(usernameStartIndex, usernameEndIndex)), new Text(value.toString()));
  		}
  		
  	}
  
  	public static class Reduce extends Reducer<Text, Text, Text, Text> {
  
  		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
  			StringBuilder builder = new StringBuilder();
  			for (Text t : values) {
  				builder.append(t+"\n");
  			}
  			context.write(key, new Text(builder.toString()));
  		}
  	}
  
  	public static void main(String[] args) throws Exception {
  		Configuration conf = new Configuration();
  		@SuppressWarnings("deprecation")
  		Job job = new Job(conf, "MessageGrouper");
  
  		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);
  	}
  
  }