Source: http://codingjunkie.net/order-inversion/

MapReduce Algorithms – Order Inversion

This post is another segment in the series presenting MapReduce algorithms as found in the Data-Intensive Text Processing with MapReducebook. Previous installments are Local Aggregation, Local Aggregation PartII and Creating a Co-Occurrence Matrix. This time we will discuss the order inversion pattern. The order inversion pattern exploits the sorting phase of MapReduce to push data needed for calculations to the reducer ahead of the data that will be manipulated.. Before you dismiss this as an edge condition for MapReduce, I urge you to read on as we will discuss how to use sorting to our advantage and cover using a custom partitioner, both of which are useful tools to have available. Although many MapReduce programs are written at a higher level abstraction i.e Hive or Pig, it’s still helpful to have an understanding of what’s going on at a lower level. The order inversion pattern is found in chapter 3 of Data-Intensive Text Processing with MapReduce book. To illustrate the order inversion pattern we will be using the Pairs approach from the co-occurrence matrix pattern. When creating the co-occurrence matrix, we track the total counts of when words appear together. At a high level we take the Pairs approach and add a small twist, in addition to having the mapper emit a word pair such as (“foo”,”bar”) we will emit an additional word pair of (“foo”,”*”) and will do so for every word pair so we can easily achieve a total count for how often the left most word appears, and use that count to calculate our relative frequencies. This approach raised two specific problems. First we need to find a way to ensure word pairs (“foo”,”*”) arrive at the reducer first. Secondly we need to make sure all word pairs with the same left word arrive at the same reducer. Before we solve those problems, let’s take a look at our mapper code.

Mapper Code

First we need to modify our mapper from the Pairs approach. At the bottom of each loop after we have emitted all the word pairs for a particular word, we will emit the special token WordPair(“word”,”*”) along with the count of times the word on the left was found.

1	public class PairsRelativeOccurrenceMapper extends Mapper<LongWritable, Text, WordPair, IntWritable> {

2	private WordPair wordPair = new WordPair();

3	private IntWritable ONE = new IntWritable(1);

4	private IntWritable totalCount = new IntWritable();

5

6

@Override

7	protected void map(LongWritable key, Text value, Context context) throwsIOException, InterruptedException {

8	int neighbors = context.getConfiguration().getInt("neighbors", 2);

9	String[] tokens = value.toString().split("\\s+");

10	if (tokens.length > 1) {

11	for (int i = 0; i < tokens.length; i++) {

12	tokens[i] = tokens[i].replaceAll("\\W+","");

13

14	if(tokens[i].equals("")){

15

continue;

16

}

17

18	wordPair.setWord(tokens[i]);

19

20	int start = (i - neighbors < 0) ? 0 : i - neighbors;

21	int end = (i + neighbors >= tokens.length) ? tokens.length - 1: i + neighbors;

22	for (int j = start; j <= end; j++) {

23	if (j == i) continue;

24	wordPair.setNeighbor(tokens[j].replaceAll("\\W",""));

25	context.write(wordPair, ONE);

26

}

27	wordPair.setNeighbor("*");

28	totalCount.set(end - start);

29	context.write(wordPair, totalCount);

30

}

31

}

32

}

33

}

Now that we’ve generated a way to track the total numbers of times a particular word has been encountered, we need to make sure those special characters reach the reducer first so a total can be tallied to calculate the relative frequencies. We will have the sorting phase of the MapReduce process handle this for us by modifying the compareTo method on the WordPair object.

Modified Sorting

We modify the compareTo method on the WordPair class so when a “*” caracter is encountered on the right that particular object is pushed to the top.

1

@Override

2	public int compareTo(WordPair other) {

3	int returnVal = this.word.compareTo(other.getWord());

4	if(returnVal != 0){

5	return returnVal;

6

}

7	if(this.neighbor.toString().equals("*")){

8	return -1;

9	}else if(other.getNeighbor().toString().equals("*")){

10

return 1;

11

}

12	return this.neighbor.compareTo(other.getNeighbor());

13

}

By modifying the compareTo method we now are guaranteed that any WordPair with the special character will be sorted to the top and arrive at the reducer first. This leads to our second specialization, how can we guarantee that all WordPair objects with a given left word will be sent to the same reducer? The answer is to create a custom partitioner.

Custom Partitioner

Intermediate keys are shuffled to reducers by calculating the hashcode of the key modulo the number of reducers. But our WordPair objects contain two words, so taking the hashcode of the entire object clearly won’t work. We need to wright a custom Partitioner that only takes into consideration the left word when it comes to determining which reducer to send the output to.

1	public class WordPairPartitioner extends Partitioner<WordPair,IntWritable> {

2

3

@Override

4	public int getPartition(WordPair wordPair, IntWritable intWritable, intnumPartitions) {

5	return wordPair.getWord().hashCode() % numPartitions;

6

}

7

}

Now we are guaranteed that all of the WordPair objects with the same left word are sent to the same reducer. All that is left is to construct a reducer to take advantage of the format of the data being sent.

Reducer

Building the reducer for the inverted order inversion pattern is straight forward. It will involve keeping a counter variable and a “current” word variable. The reducer will check the input key WordPair for the special character “*” on the right. If the word on the left is not equal to the “current” word we will re-set the counter and sum all of the values to obtain a total number of times the given current word was observed. We will now process the next WordPair objects, sum the counts and divide by our counter variable to obtain a relative frequency. This process will continue until another special character is encountered and the process starts over.

1	public class PairsRelativeOccurrenceReducer extends Reducer<WordPair, IntWritable, WordPair, DoubleWritable> {

2	private DoubleWritable totalCount = new DoubleWritable();

3	private DoubleWritable relativeCount = new DoubleWritable();

4	private Text currentWord = new Text("NOT_SET");

5	private Text flag = new Text("*");

6

7

@Override

8	protected void reduce(WordPair key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

9	if (key.getNeighbor().equals(flag)) {

10	if (key.getWord().equals(currentWord)) {

11	totalCount.set(totalCount.get() + getTotalCount(values));

12

} else {

13	currentWord.set(key.getWord());

14	totalCount.set(0);

15	totalCount.set(getTotalCount(values));

16

}

17

} else {

18	int count = getTotalCount(values);

19	relativeCount.set((double) count / totalCount.get());

20	context.write(key, relativeCount);

21

}

22

}

23	private int getTotalCount(Iterable<IntWritable> values) {

24	int count = 0;

25	for (IntWritable value : values) {

26	count += value.get();

27

}

28	return count;

29

}

30

}

By manipulating the sort order and creating a custom partitioner, we have been able to send data to a reducer needed for a calculation, before the data needed for those calculation arrive. Although not shown here, a combiner was used to run the MapReduce job. This approach is also a good candidate for the “in-mapper” combining pattern.

Example & Results

Given that the holidays are upon us, I felt it was timely to run an example of the order inversion pattern against the novel “A Christmas Carol” by Charles Dickens. I know it’s corny, but it serves the purpose.

1	new-host-2:sbin bbejeck$ hdfs dfs -cat relative/part* | grep Humbug

2	{word=[Humbug] neighbor=[Scrooge]}  0.2222222222222222

3	{word=[Humbug] neighbor=[creation]} 0.1111111111111111

4	{word=[Humbug] neighbor=[own]}  0.1111111111111111

5	{word=[Humbug] neighbor=[said]} 0.2222222222222222

6	{word=[Humbug] neighbor=[say]}  0.1111111111111111

7	{word=[Humbug] neighbor=[to]}   0.1111111111111111

8	{word=[Humbug] neighbor=[with]} 0.1111111111111111

9	{word=[Scrooge] neighbor=[Humbug]}  0.0020833333333333333

10	{word=[creation] neighbor=[Humbug]} 0.1

11	{word=[own] neighbor=[Humbug]}  0.006097560975609756

12	{word=[said] neighbor=[Humbug]} 0.0026246719160104987

13	{word=[say] neighbor=[Humbug]}  0.010526315789473684

14	{word=[to] neighbor=[Humbug]}   3.97456279809221E-4

15	{word=[with] neighbor=[Humbug]} 9.372071227741331E-4

Conclusion

While calculating relative word occurrence frequencies probably is not a common task, we have been able to demonstrate useful examples of sorting and using a custom partitioner, which are good tools to have at your disposal when building MapReduce programs. As stated before, even if most of your MapReduce is written at higher level of abstraction like Hive or Pig, it’s still instructive to have an understanding of what is going on under the hood. Thanks for your time.