TeraValidate.java
上传用户:quxuerui
上传日期:2018-01-08
资源大小:41811k
文件大小:5k
源码类别:

网格计算

开发平台:

Java

  1. /**
  2.  * Licensed to the Apache Software Foundation (ASF) under one
  3.  * or more contributor license agreements.  See the NOTICE file
  4.  * distributed with this work for additional information
  5.  * regarding copyright ownership.  The ASF licenses this file
  6.  * to you under the Apache License, Version 2.0 (the
  7.  * "License"); you may not use this file except in compliance
  8.  * with the License.  You may obtain a copy of the License at
  9.  *
  10.  *     http://www.apache.org/licenses/LICENSE-2.0
  11.  *
  12.  * Unless required by applicable law or agreed to in writing, software
  13.  * distributed under the License is distributed on an "AS IS" BASIS,
  14.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15.  * See the License for the specific language governing permissions and
  16.  * limitations under the License.
  17.  */
  18. package org.apache.hadoop.examples.terasort;
  19. import java.io.IOException;
  20. import java.util.Iterator;
  21. import org.apache.hadoop.conf.Configured;
  22. import org.apache.hadoop.fs.Path;
  23. import org.apache.hadoop.io.Text;
  24. import org.apache.hadoop.mapred.FileOutputFormat;
  25. import org.apache.hadoop.mapred.FileSplit;
  26. import org.apache.hadoop.mapred.JobClient;
  27. import org.apache.hadoop.mapred.JobConf;
  28. import org.apache.hadoop.mapred.MapReduceBase;
  29. import org.apache.hadoop.mapred.Mapper;
  30. import org.apache.hadoop.mapred.OutputCollector;
  31. import org.apache.hadoop.mapred.Reducer;
  32. import org.apache.hadoop.mapred.Reporter;
  33. import org.apache.hadoop.util.Tool;
  34. import org.apache.hadoop.util.ToolRunner;
  35. /**
  36.  * Generate 1 mapper per a file that checks to make sure the keys
  37.  * are sorted within each file. The mapper also generates 
  38.  * "$file:begin", first key and "$file:end", last key. The reduce verifies that
  39.  * all of the start/end items are in order.
  40.  * Any output from the reduce is problem report.
  41.  * <p>
  42.  * To run the program: 
  43.  * <b>bin/hadoop jar hadoop-*-examples.jar teravalidate out-dir report-dir</b>
  44.  * <p>
  45.  * If there is any output, something is wrong and the output of the reduce
  46.  * will have the problem report.
  47.  */
  48. public class TeraValidate extends Configured implements Tool {
  49.   private static final Text error = new Text("error");
  50.   static class ValidateMapper extends MapReduceBase 
  51.       implements Mapper<Text,Text,Text,Text> {
  52.     private Text lastKey;
  53.     private OutputCollector<Text,Text> output;
  54.     private String filename;
  55.     
  56.     /**
  57.      * Get the final part of the input name
  58.      * @param split the input split
  59.      * @return the "part-00000" for the input
  60.      */
  61.     private String getFilename(FileSplit split) {
  62.       return split.getPath().getName();
  63.     }
  64.     public void map(Text key, Text value, OutputCollector<Text,Text> output,
  65.                     Reporter reporter) throws IOException {
  66.       if (lastKey == null) {
  67.         filename = getFilename((FileSplit) reporter.getInputSplit());
  68.         output.collect(new Text(filename + ":begin"), key);
  69.         lastKey = new Text();
  70.         this.output = output;
  71.       } else {
  72.         if (key.compareTo(lastKey) < 0) {
  73.           output.collect(error, new Text("misorder in " + filename + 
  74.                                          " last: '" + lastKey + 
  75.                                          "' current: '" + key + "'"));
  76.         }
  77.       }
  78.       lastKey.set(key);
  79.     }
  80.     
  81.     public void close() throws IOException {
  82.       if (lastKey != null) {
  83.         output.collect(new Text(filename + ":end"), lastKey);
  84.       }
  85.     }
  86.   }
  87.   /**
  88.    * Check the boundaries between the output files by making sure that the
  89.    * boundary keys are always increasing.
  90.    * Also passes any error reports along intact.
  91.    */
  92.   static class ValidateReducer extends MapReduceBase 
  93.       implements Reducer<Text,Text,Text,Text> {
  94.     private boolean firstKey = true;
  95.     private Text lastKey = new Text();
  96.     private Text lastValue = new Text();
  97.     public void reduce(Text key, Iterator<Text> values,
  98.                        OutputCollector<Text, Text> output, 
  99.                        Reporter reporter) throws IOException {
  100.       if (error.equals(key)) {
  101.         while(values.hasNext()) {
  102.           output.collect(key, values.next());
  103.         }
  104.       } else {
  105.         Text value = values.next();
  106.         if (firstKey) {
  107.           firstKey = false;
  108.         } else {
  109.           if (value.compareTo(lastValue) < 0) {
  110.             output.collect(error, 
  111.                            new Text("misordered keys last: " + 
  112.                                     lastKey + " '" + lastValue +
  113.                                     "' current: " + key + " '" + value + "'"));
  114.           }
  115.         }
  116.         lastKey.set(key);
  117.         lastValue.set(value);
  118.       }
  119.     }
  120.     
  121.   }
  122.   public int run(String[] args) throws Exception {
  123.     JobConf job = (JobConf) getConf();
  124.     TeraInputFormat.setInputPaths(job, new Path(args[0]));
  125.     FileOutputFormat.setOutputPath(job, new Path(args[1]));
  126.     job.setJobName("TeraValidate");
  127.     job.setJarByClass(TeraValidate.class);
  128.     job.setMapperClass(ValidateMapper.class);
  129.     job.setReducerClass(ValidateReducer.class);
  130.     job.setOutputKeyClass(Text.class);
  131.     job.setOutputValueClass(Text.class);
  132.     // force a single reducer
  133.     job.setNumReduceTasks(1);
  134.     // force a single split 
  135.     job.setLong("mapred.min.split.size", Long.MAX_VALUE);
  136.     job.setInputFormat(TeraInputFormat.class);
  137.     JobClient.runJob(job);
  138.     return 0;
  139.   }
  140.   /**
  141.    * @param args
  142.    */
  143.   public static void main(String[] args) throws Exception {
  144.     int res = ToolRunner.run(new JobConf(), new TeraValidate(), args);
  145.     System.exit(res);
  146.   }
  147. }