Sorted word count using Hadoop MapReduce

In simple word count map reduce program the output we get is sorted by words. Sample output can be :
Apple 1
Boy 30
Cat 2
Frog 20
Zebra 1
If you want output to be sorted on the basis of number of occrance of words, i.e in below format
1 Apple
1 Zebra
2 Cat
20 Frog
30 Boy
You can create another MR program using below mapper and reducer where the input will be the output got from simple word count program.

class Map1 extends MapReduceBase implements Mapper<Object, Text, IntWritable, Text>{    public void map(Object key, Text value, OutputCollector<IntWritable, Text> collector, Reporter arg3) throws IOException     {        String line = value.toString();        StringTokenizer stringTokenizer = new StringTokenizer(line);        {            int number = 999;             String word = "empty";            if(stringTokenizer.hasMoreTokens())            {                String str0= stringTokenizer.nextToken();                word = str0.trim();            }            if(stringTokenizer.hasMoreElements())            {                String str1 = stringTokenizer.nextToken();                number = Integer.parseInt(str1.trim());            }            collector.collect(new IntWritable(number), new Text(word));        }    }}class Reduce1 extends MapReduceBase implements Reducer<IntWritable, Text, IntWritable, Text>{    public void reduce(IntWritable key, Iterator<Text> values, OutputCollector<IntWritable, Text> arg2, Reporter arg3) throws IOException    {        while((values.hasNext()))        {            arg2.collect(key, values.next());        }    }}

The output from the Hadoop MapReduce wordcount example is sorted by the key. So the output should be in alphabetical order.

With Hadoop you can create your own key objects that implement the WritableComparable interface allowing you to override the compareTo method. This allows you to control the sort order.

To create an output that is sorted by the number of occurances you would probably have to add another MapReduce job to process the output from the first as you have said. This second job would be very simple, maybe not even requiring a reduce phase. You would just need to implement your own Writable key object to wrap the word and its frequency. A custom writable looks something like this:

 public class MyWritableComparable implements WritableComparable {       // Some data       private int counter;       private long timestamp;       public void write(DataOutput out) throws IOException {         out.writeInt(counter);         out.writeLong(timestamp);       }       public void readFields(DataInput in) throws IOException {         counter = in.readInt();         timestamp = in.readLong();       }       public int compareTo(MyWritableComparable w) {         int thisValue = this.value;         int thatValue = ((IntWritable)o).value;         return (thisValue < thatValue ? -1 : (thisValue==thatValue ? 0 : 1));       }     }

I grabbed this example from here.

You should probably override hashCode, equals and toString as well.

In Hadoop sorting is done between the Map and the Reduce phases. One approach to sort by word occurance would be to use a custom group comparator that doesn't group anything; therefore, every call to reduce is just the key and one value.

public class Program {   public static void main( String[] args) {      conf.setOutputKeyClass( IntWritable.class);      conf.setOutputValueClass( Text.clss);      conf.setMapperClass( Map.class);      conf.setReducerClass( IdentityReducer.class);      conf.setOutputValueGroupingComparator( GroupComparator.class);         conf.setNumReduceTasks( 1);      JobClient.runJob( conf);   }}public class Map extends MapReduceBase implements Mapper<Text,IntWritable,IntWritable,Text> {   public void map( Text key, IntWritable value, OutputCollector<IntWritable,Text>, Reporter reporter) {       output.collect( value, key);   }}public class GroupComaprator extends WritableComparator {    protected GroupComparator() {        super( IntWritable.class, true);    }    public int compare( WritableComparable w1, WritableComparable w2) {        return -1;    }}