{"id":2045,"date":"2012-10-17T10:00:00","date_gmt":"2012-10-17T10:00:00","guid":{"rendered":"http:\/\/www.javacodegeeks.com\/2012\/10\/mapreduce-working-through-data-intensive-text-processing-local-aggregation-part-ii.html"},"modified":"2012-10-28T23:34:59","modified_gmt":"2012-10-28T23:34:59","slug":"mapreduce-working-through-data-2","status":"publish","type":"post","link":"https:\/\/www.javacodegeeks.com\/2012\/10\/mapreduce-working-through-data-2.html","title":{"rendered":"MapReduce: Working Through Data-Intensive Text Processing – Local Aggregation Part II"},"content":{"rendered":"
This post continues with the series on implementing algorithms found in the Data Intensive Processing with MapReduce<\/a> book. Part one can be found here<\/a>. In the previous post, we discussed using the technique of local aggregation as a means of reducing the amount of data shuffled and transferred across the network. Reducing the amount of data transferred is one of the top ways to improve the efficiency of a MapReduce job. A word-count MapReduce job was used to demonstrate local aggregation. Since the results only require a total count, we could re-use the same reducer for our combiner as changing the order or groupings of the addends will not affect the sum.<\/p>\n

But what if you wanted an average<\/i>? Then the same approach would not work because calculating an average of averages is not equal to the average of the original set of numbers. With a little bit of insight though, we can still use local aggregation. For these examples we will be using a sample of the NCDC weather dataset<\/a> used in Hadoop the Definitive Guide<\/a> book. We will calculate the average temperature for each month in the year 1901. The averages algorithm for the combiner and the in-mapper combining option can be found in chapter 3.1.3 of Data-Intensive Processing with MapReduce. <\/p>\n

One Size Does Not Fit All<\/strong><\/p>\n

Last time we described two approaches for reducing data in a MapReduce job, Hadoop Combiners and the in-mapper combining approach. Combiners are considered an optimization by the Hadoop framework and there are no guarantees on how many times they will be called, if at all. As a result, mappers must emit data in the form expected by the reducers so if combiners aren\u2019t involved, the final result is not changed. To adjust for calculating averages, we need to go back to the mapper and change it\u2019s output.
\n
\n<\/strong>Mapper Changes<\/strong> <\/p>\n

In the word-count example, the non-optimized mapper simply emitted the word and the count of 1. The combiner and in-mapper combining mapper optimized this output by keeping each word as a key in a hash map with the total count as the value. Each time a word was seen the count was incremented by 1. With this setup, if the combiner was not called, the reducer would receive the word as a key and a long list of 1?s to add together, resulting in the same output (of course using the in-mapper combining mapper avoided this issue because it\u2019s guaranteed to combine results as it\u2019s part of the mapper code). To compute an average, we will have our base mapper emit a string key (the year and month of the weather observation concatenated together) and a custom writable<\/a> object, called TemperatureAveragingPair. The TemperatureAveragingPair<\/a> object will contain two numbers (IntWritables), the temperature taken and a count of one. We will take the MaximumTemperatureMapper from Hadoop: The Definitive Guide and use it as inspiration for creating an AverageTemperatureMapper: <\/p>\n

public class AverageTemperatureMapper extends Mapper<LongWritable, Text, Text, TemperatureAveragingPair> {\r\n \/\/sample line of weather data\r\n \/\/0029029070999991901010106004+64333+023450FM-12+000599999V0202701N015919999999N0000001N9-00781+99999102001ADDGF10899199999999999\r\n\r\n\r\n    private Text outText = new Text();\r\n    private TemperatureAveragingPair pair = new TemperatureAveragingPair();\r\n    private static final int MISSING = 9999;\r\n\r\n    @Override\r\n    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {\r\n        String line = value.toString();\r\n        String yearMonth = line.substring(15, 21);\r\n\r\n        int tempStartPosition = 87;\r\n\r\n        if (line.charAt(tempStartPosition) == '+') {\r\n            tempStartPosition += 1;\r\n        }\r\n\r\n        int temp = Integer.parseInt(line.substring(tempStartPosition, 92));\r\n\r\n        if (temp != MISSING) {\r\n            outText.set(yearMonth);\r\n            pair.set(temp, 1);\r\n            context.write(outText, pair);\r\n        }\r\n    }\r\n}<\/pre>\n
By having the mapper output a key and TemperatureAveragingPair object our MapReduce program is guaranteed to have the correct results regardless if the combiner is called.
\n
\n<\/strong>Combiner<\/strong>
 
<\/div> <\/div><\/p>\n
We need to reduce the amount of data sent, so we will sum the temperatures, and sum the counts and store them separately. By doing so we will reduce data sent, but preserve the format needed for calculating correct averages. If\/when the combiner is called, it will take all the TemperatureAveragingPair objects passed in and emit a single TemperatureAveragingPair object for the same key, containing the summed temperatures and counts. Here is the code for the combiner:          <\/p>\n
  public class AverageTemperatureCombiner extends Reducer<Text,TemperatureAveragingPair,Text,TemperatureAveragingPair> {\r\n    private TemperatureAveragingPair pair = new TemperatureAveragingPair();\r\n\r\n    @Override\r\n    protected void reduce(Text key, Iterable<TemperatureAveragingPair> values, Context context) throws IOException, InterruptedException {\r\n        int temp = 0;\r\n        int count = 0;\r\n        for (TemperatureAveragingPair value : values) {\r\n             temp += value.getTemp().get();\r\n             count += value.getCount().get();\r\n        }\r\n        pair.set(temp,count);\r\n        context.write(key,pair);\r\n    }\r\n}<\/pre>\n
But we are really interested in being guaranteed we have reduced the amount of data sent to the reducers, so we\u2019ll have a look at how to achieve that next.
\n
\n<\/strong>In Mapper Combining Averages<\/strong>        <\/p>\n
Similar to the word-count example, for calculating averages, the in-mapper-combining mapper will use a hash map with the concatenated year+month as a key and a TemperatureAveragingPair as the value. Each time we get the same year+month combination, we\u2019ll take the pair object out of the map, add the temperature and increase the count by by one. Once the cleanup method is called we\u2019ll and emit all pairs with their respective key:         <\/p>\n
public class AverageTemperatureCombiningMapper extends Mapper<LongWritable, Text, Text, TemperatureAveragingPair> {\r\n \/\/sample line of weather data\r\n \/\/0029029070999991901010106004+64333+023450FM-12+000599999V0202701N015919999999N0000001N9-00781+99999102001ADDGF10899199999999999\r\n\r\n\r\n    private static final int MISSING = 9999;\r\n    private Map<String,TemperatureAveragingPair> pairMap = new HashMap<String,TemperatureAveragingPair>();\r\n\r\n\r\n    @Override\r\n    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {\r\n        String line = value.toString();\r\n        String yearMonth = line.substring(15, 21);\r\n\r\n        int tempStartPosition = 87;\r\n\r\n        if (line.charAt(tempStartPosition) == '+') {\r\n            tempStartPosition += 1;\r\n        }\r\n\r\n        int temp = Integer.parseInt(line.substring(tempStartPosition, 92));\r\n\r\n        if (temp != MISSING) {\r\n            TemperatureAveragingPair pair = pairMap.get(yearMonth);\r\n            if(pair == null){\r\n                pair = new TemperatureAveragingPair();\r\n                pairMap.put(yearMonth,pair);\r\n            }\r\n            int temps = pair.getTemp().get() + temp;\r\n            int count = pair.getCount().get() + 1;\r\n            pair.set(temps,count);\r\n        }\r\n    }\r\n\r\n\r\n    @Override\r\n    protected void cleanup(Context context) throws IOException, InterruptedException {\r\n        Set<String> keys = pairMap.keySet();\r\n        Text keyText = new Text();\r\n        for (String key : keys) {\r\n             keyText.set(key);\r\n             context.write(keyText,pairMap.get(key));\r\n        }\r\n    }\r\n}<\/pre>\n
By following the same pattern of keeping track of data between map calls, we can achieve reliable data reduction by implementing an in-mapper combining strategy. The same caveats apply for keeping state across all calls to the mapper, but looking at the gains that can be made in processing efficiency, using this approach merits some consideration.
\n
\n<\/strong>Reducer<\/strong>        <\/p>\n
At this point, writing our reducer is easy, take a list of pairs for each key, sum all the temperatures and counts then divide the sum of the temperatures by the sum of the counts.         <\/p>\n
public class AverageTemperatureReducer extends Reducer<Text, TemperatureAveragingPair, Text, IntWritable> {\r\n    private IntWritable average = new IntWritable();\r\n\r\n    @Override\r\n    protected void reduce(Text key, Iterable<TemperatureAveragingPair> values, Context context) throws IOException, InterruptedException {\r\n        int temp = 0;\r\n        int count = 0;\r\n        for (TemperatureAveragingPair pair : values) {\r\n            temp += pair.getTemp().get();\r\n            count += pair.getCount().get();\r\n        }\r\n        average.set(temp \/ count);\r\n        context.write(key, average);\r\n    }\r\n}\r\n<\/pre>\n

\n<\/strong>Results<\/strong>        <\/p>\n
The results are predictable with the combiner and in-mapper-combining mapper options showing substantially reduced data output.
\nNon-Optimized Mapper Option:         <\/p>\n
12\/10\/10 23:05:28 INFO mapred.JobClient:     Reduce input groups=12\r\n12\/10\/10 23:05:28 INFO mapred.JobClient:     Combine output records=0\r\n12\/10\/10 23:05:28 INFO mapred.JobClient:     Map input records=6565\r\n12\/10\/10 23:05:28 INFO mapred.JobClient:     Reduce shuffle bytes=111594\r\n12\/10\/10 23:05:28 INFO mapred.JobClient:     Reduce output records=12\r\n12\/10\/10 23:05:28 INFO mapred.JobClient:     Spilled Records=13128\r\n12\/10\/10 23:05:28 INFO mapred.JobClient:     Map output bytes=98460\r\n12\/10\/10 23:05:28 INFO mapred.JobClient:     Total committed heap usage (bytes)=269619200\r\n12\/10\/10 23:05:28 INFO mapred.JobClient:     Combine input records=0\r\n12\/10\/10 23:05:28 INFO mapred.JobClient:     Map output records=6564\r\n12\/10\/10 23:05:28 INFO mapred.JobClient:     SPLIT_RAW_BYTES=108\r\n12\/10\/10 23:05:28 INFO mapred.JobClient:     Reduce input records=6564<\/pre>\n
Combiner Option:         <\/p>\n
12\/10\/10 23:07:19 INFO mapred.JobClient:     Reduce input groups=12\r\n12\/10\/10 23:07:19 INFO mapred.JobClient:     Combine output records=12\r\n12\/10\/10 23:07:19 INFO mapred.JobClient:     Map input records=6565\r\n12\/10\/10 23:07:19 INFO mapred.JobClient:     Reduce shuffle bytes=210\r\n12\/10\/10 23:07:19 INFO mapred.JobClient:     Reduce output records=12\r\n12\/10\/10 23:07:19 INFO mapred.JobClient:     Spilled Records=24\r\n12\/10\/10 23:07:19 INFO mapred.JobClient:     Map output bytes=98460\r\n12\/10\/10 23:07:19 INFO mapred.JobClient:     Total committed heap usage (bytes)=269619200\r\n12\/10\/10 23:07:19 INFO mapred.JobClient:     Combine input records=6564\r\n12\/10\/10 23:07:19 INFO mapred.JobClient:     Map output records=6564\r\n12\/10\/10 23:07:19 INFO mapred.JobClient:     SPLIT_RAW_BYTES=108\r\n12\/10\/10 23:07:19 INFO mapred.JobClient:     Reduce input records=12<\/pre>\n
In-Mapper-Combining Option:         <\/p>\n
12\/10\/10 23:09:09 INFO mapred.JobClient:     Reduce input groups=12\r\n12\/10\/10 23:09:09 INFO mapred.JobClient:     Combine output records=0\r\n12\/10\/10 23:09:09 INFO mapred.JobClient:     Map input records=6565\r\n12\/10\/10 23:09:09 INFO mapred.JobClient:     Reduce shuffle bytes=210\r\n12\/10\/10 23:09:09 INFO mapred.JobClient:     Reduce output records=12\r\n12\/10\/10 23:09:09 INFO mapred.JobClient:     Spilled Records=24\r\n12\/10\/10 23:09:09 INFO mapred.JobClient:     Map output bytes=180\r\n12\/10\/10 23:09:09 INFO mapred.JobClient:     Total committed heap usage (bytes)=269619200\r\n12\/10\/10 23:09:09 INFO mapred.JobClient:     Combine input records=0\r\n12\/10\/10 23:09:09 INFO mapred.JobClient:     Map output records=12\r\n12\/10\/10 23:09:09 INFO mapred.JobClient:     SPLIT_RAW_BYTES=108\r\n12\/10\/10 23:09:09 INFO mapred.JobClient:     Reduce input records=12<\/pre>\n
Calculated Results:
\n(NOTE: the temperatures in the sample file are in Celsius * 10)         <\/p>\n\n\n\n\n
Non-Optimized<\/td>\nCombiner      <\/td>\nIn-Mapper-Combiner Mapper<\/td>\n<\/tr>\n
190101 -25
\n190102 -91
\n190103 -49
\n190104 22
\n190105 76
\n190106 146
\n190107 192
\n190108 170
\n190109 114
\n190110 86
\n190111 -16
\n190112 -77 <\/td>\n		190101 -25
\n190102 -91
\n190103 -49
\n190104 22
\n190105 76
\n190106 146
\n190107 192
\n190108 170
\n190109 114
\n190110 86
\n190111 -16
\n190112 -77 <\/td>\n		190101 -25
\n190102 -91
\n190103 -49
\n190104 22
\n190105 76
\n190106 146
\n190107 192
\n190108 170
\n190109 114
\n190110 86
\n190111 -16
\n190112 -77 <\/td>\n<\/tr>\n<\/tbody>\n<\/table>\n

\n<\/strong>Conclusion<\/strong>        <\/p>\n
We have covered local aggregation for both the simple case where one could reuse the reducer as a combiner and a more complicated case where some insight on how to structure the data while still gaining the benefits of locally aggregating data for increase processing efficiency.
\n
\nFurther Reading<\/strong>
\n
\n<\/strong><\/p>\n