Get AvroSpecificMaxTemperature test passing.

tomwhite · Jan 15, 2012 · 2e2ca8b · 2e2ca8b
1 parent ca7ee34
commit 2e2ca8b
Show file tree

Hide file tree

Showing 4 changed files with 29 additions and 6 deletions.
diff --git a/ch04-avro/src/main/examples/AvroSpecificMaxTemperature/input.txt b/ch04-avro/src/main/examples/AvroSpecificMaxTemperature/input.txt
@@ -0,0 +1 @@
+hadoop jar avro-examples.jar AvroSpecificMaxTemperature input/ncdc/sample.txt output
diff --git a/ch04-avro/src/main/examples/AvroSpecificMaxTemperature/output/part-00000.avro b/ch04-avro/src/main/examples/AvroSpecificMaxTemperature/output/part-00000.avro
diff --git a/ch04-avro/src/main/java/AvroSpecificMaxTemperature.java b/ch04-avro/src/main/java/AvroSpecificMaxTemperature.java
@@ -18,6 +18,8 @@
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
 
+import specific.WeatherRecord;
+
 public class AvroSpecificMaxTemperature extends Configured implements Tool {
 
   public static class MaxTemperatureMapper
@@ -54,15 +56,33 @@ public void reduce(Integer key, Iterable<WeatherRecord> values,
       }
       collector.collect(max);
     }
-    private WeatherRecord newWeatherRecord(WeatherRecord value) {
-      WeatherRecord record = new WeatherRecord();
-      record.year = value.year;
-      record.temperature = value.temperature;
-      record.stationId = value.stationId;
-      return record;
+  }
+
+  public static class MaxTemperatureCombiner extends
+      AvroReducer<Integer, WeatherRecord, Pair<Integer, WeatherRecord>> {
+
+    @Override
+    public void reduce(Integer key, Iterable<WeatherRecord> values,
+        AvroCollector<Pair<Integer, WeatherRecord>> collector,
+        Reporter reporter) throws IOException {
+      WeatherRecord max = null;
+      for (WeatherRecord value : values) {
+        if (max == null || value.temperature > max.temperature) {
+          max = newWeatherRecord(value);
+        }
+      }
+      collector.collect(new Pair<Integer, WeatherRecord>(key, max));
     }
   }
 
+  private static WeatherRecord newWeatherRecord(WeatherRecord value) {
+    WeatherRecord record = new WeatherRecord();
+    record.year = value.year;
+    record.temperature = value.temperature;
+    record.stationId = value.stationId;
+    return record;
+  }
+
   @Override
   public int run(String[] args) throws Exception {
     if (args.length != 2) {
@@ -86,6 +106,7 @@ public int run(String[] args) throws Exception {
     conf.setInputFormat(AvroUtf8InputFormat.class);
 
     AvroJob.setMapperClass(conf, MaxTemperatureMapper.class);
+    AvroJob.setCombinerClass(conf, MaxTemperatureCombiner.class);
     AvroJob.setReducerClass(conf, MaxTemperatureReducer.class);
 
     JobClient.runJob(conf);

diff --git a/ch04-avro/src/main/resources/WeatherRecord.avsc b/ch04-avro/src/main/resources/WeatherRecord.avsc
@@ -1,6 +1,7 @@
 {
   "type": "record",
   "name": "WeatherRecord",
+  "namespace": "specific",
   "doc": "A weather reading.",
   "fields": [
     {"name": "year", "type": "int"},