Map Reduce Secondary Sort Does It All

I came across a question in Stack Overflow recently related to calculating a web chat room statistics using Hadoop Map Reduce. The answer to the question was begging for a solution based map reduce secondary sort. I will provide  details, along with code snippet,  to complement my answer to the question.

The Problem

The data consists of a time stamp, chat room zone and number of users. The data is logged once per minute.  The goal is to find maximum number of users per day for all zones. The data looks like as below

timestamp; zone; value
2011-01-01 00:00:00; 1; 10
2011-01-01 00:00:00; 2; 22
2011-01-01 00:01:00; 1; 11

The Solution

The brute force solution is for the map to emit zone and day as the key and the number of users as the value. The reducer will receiver all the user counts and it will find the maximum iterating through the value list. But we can do better than that, thanks to Hadoop secondary sort. To reiterate my answer in stack overflow, the solution based on secondary sort, consists of the following steps.

1. Define the key as concatenation of zone, timestamp and the value i.e, zone:yyyy-mm-dd:value You don’t even need to emit any value from the mapper. NullWritable is good enough for the value.
2. Implement key comparator such that zone:yyyy-mm-dd part of the key is ordered ascending and the value part is ordered descending. This will ensure that for all keys for given zone:yyyy-mm-dd, the first key in the group will have the highest value.
3. Define partitioner and grouping comparator of the composite key based on the zone and day part of the key only i.e. zone:yyyy-mm-dd.
4. The reducer input will contain the first key for a key group, which will contain zone, day and the max value for that zone, day combination. The value part of the reducer input will be a list of NullWritable, which can be ignored.

Implementation

The key emitted by the mapper is a composite key. The zone and day part of the key is the natural key. The natural key is used by the partitioner and the group comparator. When we tag a value to the natural key, we get a composite key, which gives us sorting ability for values, because the value become part of the key sorting

Chat zone

Date(Year,month and day)

User count(per minute)

The mapper  emits the key as defined in step1 and NullWritable as values. The mapper output will look like this.

2:2011-01-01:23 -> null
2:2011-01-05:17 -> null

The key comparator implementation as defined in step 2 is as follows. The composite key is used by the key comparator.

public static class KeyComprator extends WritableComparator {

 protected KeyComprator() {
 super(Text.class, true);
 }

 @Override
 public int compare(WritableComparable w1, WritableComparable w2) {

//ascending zone and day

Text t1 = (Text) w1;
 Text t2 = (Text) w2;
 String[] t1Items = t1.toString().split(":");
 String[] t2Items = t2.toString().split(":");
 String t1Base = t1Items[0] + ":" + t1Items[1] + ":";
 String t2Base = t2Items[0] + ":" + t2Items[1] + ":";
 int comp = t1Base.compareTo(t2Base);

//descending value
 if (comp == 0) {
 comp = t2Items[2].compareTo(t1Items[2]);
 }

 return comp;

 }
 }

The group comparator as mentioned in step 3, is needed because we want all the keys with the same zone and day to be grouped together. Each invocation of the the reducer will be for one group of keys. The group comparator implementation is as follows

public static class GroupComprator extends WritableComparator {

 protected GroupComprator() {
 super(Text.class, true);
 }

 @Override
 public int compare(WritableComparable w1, WritableComparable w2) {

//consider only zone and day part of the key
 Text t1 = (Text) w1;
 Text t2 = (Text) w2;
 String[] t1Items = t1.toString().split(":");
 String[] t2Items = t2.toString().split(":");
 String t1Base = t1Items[0] + ":" + t1Items[1] + ":";
 String t2Base = t2Items[0] + ":" + t2Items[1] + ":";
 int comp = t1Base.compareTo(t2Base);

 return comp;

 }
 }

By default, each key maps to a separate group. Our custom group comparator, maps multiple keys to one group.  A reducer gets called once per group. When multiple keys are mapped to one group, Hadoop sends the first key of the group to the reducer. Since, in our case the keys within a group are sorted, we get the desired key in the reducer input.

In a real Hadoop cluster, there are many reducers running in different nodes. If the data for the same zone and day don’t land in the same reducer after the map reduce shuffle, we are in trouble. The way to ensure that is taking charge of defining our own partitioning logic. The implementation is as follows

public static class ZoneDayPartitioner extends Partitioner  implements Configurable{
 @Override
 public int getPartition(Text key, Text NullWritable, int numPartitions) {
//consider only zone and day part of the key
 String[] keyItems = key.toString().split(":");
 String keyBase = keyItems[0] + ":" + keyItems[1] + ":";
 int part  = keyBase.hashCode() % numPartitions;
 return part;
 }

All the reducer will have to is to simply emit key. The reducer output will consist of zone, day and corresponding maximum visitor for that zone and day. The zone and day will be sorted ascending for a particular reducer output. But they won’t be globally sorted across all reducer outputs

Wrapping up

We could get away with emitting null for value in mapper, because we were only interested in the maximum number of users for a day and we essentially carried that value in the key.

If however, we were interested in finding top two number of users  for a day, then from the mapper, we had to emit the number of users as the value, in addition to carrying it with the key. The reducer logic also would have been different. We had to iterate through the values and retain the first two value from the list. You might find this post on map reduce secondary sort interesting read.

I have provided only the key aspects of the implementation. it should be easy to fill in the gaps for the full implementation. As usual, comments are welcome.

For commercial support for any solution in my github repositories, please talk to ThirdEye Data Science Services. Support is available for Hadoop or Spark deployment on cloud including installation, configuration and testing,