[Hadoop in Action] 第6章 编程实践

                  1
                import
                 java.io.IOException;

                  2
                import
                 java.util.regex.PatternSyntaxException;

                  3
                import
                 java.util.Iterator;

                  4
                  5
                import
                 org.apache.hadoop.conf.Configuration;

                  6
                import
                 org.apache.hadoop.conf.Configured;

                  7
                import
                 org.apache.hadoop.fs.Path;

                  8
                import
                 org.apache.hadoop.io.IntWritable;

                  9
                import
                 org.apache.hadoop.io.LongWritable;

                 10
                import
                 org.apache.hadoop.io.DoubleWritable;

                 11
                import
                 org.apache.hadoop.io.Text;

                 12
                import org.apache.hadoop.mapred.*;
 13import org.apache.hadoop.util.Tool;
 14import org.apache.hadoop.util.ToolRunner;
 15 16 17publicclass AveragingWithCombiner extends Configured implements Tool {
 18 19publicstaticclass MapClass extends MapReduceBase
 20implements Mapper<LongWritable, Text, Text, Text> {
 21 22staticenum ClaimsCounters { MISSING, QUOTED };
 23 24publicvoid map(LongWritable key, Text value,
 25                         OutputCollector<Text, Text> output,
 26                         Reporter reporter) throws IOException {
 27 28             String fields[] = value.toString().split(",", -20);
 29             String country = fields[4];
 30             String numClaims = fields[8];
 31if (numClaims.length() == 0) {
 32                 reporter.incrCounter(ClaimsCounters.MISSING, 1);
 33             } elseif (numClaims.startsWith("\"")) {
 34                 reporter.incrCounter(ClaimsCounters.QUOTED, 1);
 35             } else {
 36                 output.collect(new Text(country), new Text(numClaims + ",1"));
 37            }
 38 39        }
 40    }
 41 42publicstaticclass Combine extends MapReduceBase
 43implements Reducer<Text, Text, Text, Text> {
 44 45publicvoid reduce(Text key, Iterator<Text> values,
 46                            OutputCollector<Text, Text> output,
 47                            Reporter reporter) throws IOException {
 48 49double sum = 0;
 50int count = 0;
 51while (values.hasNext()) {
 52                 String fields[] = values.next().toString().split(",");
 53                 sum += Double.parseDouble(fields[0]);
 54                 count += Integer.parseInt(fields[1]);
 55            }
 56             output.collect(key, new Text(sum + "," + count));
 57        }
 58    }
 59 60publicstaticclass Reduce extends MapReduceBase
 61implements Reducer<Text, Text, Text, DoubleWritable> {
 62 63publicvoid reduce(Text key, Iterator<Text> values,
 64                            OutputCollector<Text, DoubleWritable> output,
 65                            Reporter reporter) throws IOException {
 66 67double sum = 0;
 68int count = 0;
 69while (values.hasNext()) {
 70                 String fields[] = values.next().toString().split(",");
 71                 sum += Double.parseDouble(fields[0]);
 72                 count += Integer.parseInt(fields[1]);
 73            }
 74             output.collect(key, new DoubleWritable(sum/count));
 75        }
 76    }
 77 78publicint run(String[] args) throws Exception {
 79// Configuration processed by ToolRunner 80         Configuration conf = getConf();
 81 82// Create a JobConf using the processed conf 83         JobConf job = new JobConf(conf, AveragingWithCombiner.class);
 84 85// Process custom command-line options 86         Path in = new Path(args[0]);
 87         Path out = new Path(args[1]);
 88        FileInputFormat.setInputPaths(job, in);
 89        FileOutputFormat.setOutputPath(job, out);
 90 91// Specify various job-specific parameters      92         job.setJobName("AveragingWithCombiner");
 93         job.setMapperClass(MapClass.class);
 94         job.setCombinerClass(Combine.class);
 95         job.setReducerClass(Reduce.class);
 96 97         job.setInputFormat(TextInputFormat.class);
 98         job.setOutputFormat(TextOutputFormat.class);
 99         job.setOutputKeyClass(Text.class);
100         job.setOutputValueClass(Text.class);
101102// Submit the job, then poll for progress until the job is complete103        JobClient.runJob(job);
104105return 0;
106    }
107108publicstaticvoid main(String[] args) throws Exception {
109// Let ToolRunner handle generic command-line options 110int res = ToolRunner.run(new Configuration(), new AveragingWithCombiner(), args);
111112        System.exit(res);
113    }
114 }