Pseudo-Combiners in Cascading

In order to get maximum performance from MapReduce, you need to minimize the amount of data that you have to transfer around the network.  If nearly your entire input must be transferred from your mappers to your reducers, then you’ll be putting a great deal of stress on your disks and network.  One thing that comes highly recommended is the use of combiners, which allow for part of the reducing to be done during the map phase in cases where you are performing associative and commutative aggregations such as counting, summing, or finding the minimum or maximum.  This is especially true when you have very few group keys, which would force large numbers of tuples to be passed into a small number of reducers.

Unfortunately, while MapReduce supports combiners, Cascading does not. Instead, we decided to hack together our own solution, which we’re calling a “pseudo-combiner”.  A traditional combiner maintains a buffer of tuples and does sorting and aggregation when the buffer fills up. Our pseudo-combiner maintains a map with the group field as the key and the combiner output as the value. For every tuple, we will perform the combiner functionality when we update the entry in the map. This is better for our most common use case, which involves counting billions of values in fewer than 100 categories. Since we can very easily hold all of our categories in the map, we can ensure that we only have one output per key value from each mapper.

Our implementation uses an LRUHashMap, which is an in-house extension of LinkedHashMap.  The LRUHashMap uses the LinkedHashMap to maintain a cap on how many entries are allowed in the map and evicts the oldest key, value pairs when the map grows beyond its limit. The evicted pair is made available so that we can emit the correct output for it. When all the input tuples have been read, we merely flush all the contents of the hash and emit all necessary tuples for them.

The abstract class we’ve designed has three functions that should be implemented by every combiner:

The initialize function is called the first time we see a key, and allows us to store the initial combiner value for the key.  The update function is called whenever we see a value for a key we’ve already seen.  The current value in the map and the new tuple are passed in.  The final getTuple function is called whenever we need to decide what to emit for an entry in our map.  This occurs on eviction and when we flush the contents of the map at the end.

Thus, our use case of counting would look like this:

Our adoption of combiners gave us huge performance improvements, cutting one of our stats jobs from an hour down to around two minutes, which means we can run the stats hourly.