官网MapReduce实例代码详细批注

客官°小女子只卖身不卖艺 2022-08-07 06:43 110阅读 0赞

引言 
      
       1.本文不描述MapReduce入门知识，这类知识网上很多，请自行查阅 
      
       2.本文的实例代码来自官网 
      
       http://hadoop.apache.org/docs/current/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html 
      
       最后的WordCount v2.0，该代码相比源码中的org.apache.hadoop.examples.WordCount要复杂和完整，更适合作为MapReduce模板代码 
      
       3.本文的目的就是为开发MapReduce的同学提供一个详细注释了的模板，可以基于该模板做开发。 
     
      
     
      
    
     
      
     
      
      
       import java.io.BufferedReader;
    import java.io.FileReader;
    import java.io.IOException;
    import java.net.URI;
    import java.util.ArrayList;
    import java.util.HashSet;
    import java.util.List;
    import java.util.Set;
    import java.util.StringTokenizer;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.mapreduce.Counter;
    import org.apache.hadoop.util.GenericOptionsParser;
    import org.apache.hadoop.util.StringUtils;
    public class WordCount2 {
        // 日志组名MapCounters，日志名INPUT_WORDS
        static enum MapCounters {
            INPUT_WORDS
        }
        static enum ReduceCounters {
            OUTPUT_WORDS
        }
        
        // static enum CountersEnum { INPUT_WORDS,OUTPUT_WORDS }
        // 日志组名CountersEnum，日志名INPUT_WORDS和OUTPUT_WORDS
        
        public static class TokenizerMapper extends
                Mapper<Object, Text, Text, IntWritable> {
            private final static IntWritable one = new IntWritable(1); // map输出的value
            private Text word = new Text(); // map输出的key
            private boolean caseSensitive; // 是否大小写敏感，从配置文件中读出赋值
            private Set<String> patternsToSkip = new HashSet<String>(); // 用来保存需过滤的关键词，从配置文件中读出赋值
            private Configuration conf;
            private BufferedReader fis; // 保存文件输入流
            /**
             * 整个setup就做了两件事： 1.读取配置文件中的wordcount.case.sensitive，赋值给caseSensitive变量
             * 2.读取配置文件中的wordcount.skip.patterns，如果为true，将CacheFiles的文件都加入过滤范围
             */
            @Override
            public void setup(Context context) throws IOException,
                    InterruptedException {
                conf = context.getConfiguration();
                // getBoolean(String name, boolean defaultValue)
                // 获取name指定属性的值，如果属性没有指定，或者指定的值无效，就用defaultValue返回。
                // 属性可以在命令行中通过-Dpropretyname指定，例如 -Dwordcount.case.sensitive=true
                // 属性也可以在main函数中通过job.getConfiguration().setBoolean("wordcount.case.sensitive",
                // true)指定
                caseSensitive = conf.getBoolean("wordcount.case.sensitive", true); // 配置文件中的wordcount.case.sensitive功能是否打开
                // wordcount.skip.patterns属性的值取决于命令行参数是否有-skip，具体逻辑在main方法中
                if (conf.getBoolean("wordcount.skip.patterns", false)) { // 配置文件中的wordcount.skip.patterns功能是否打开
                    URI[] patternsURIs = Job.getInstance(conf).getCacheFiles(); // getCacheFiles()方法可以取出缓存的本地化文件，本例中在main设置
                    for (URI patternsURI : patternsURIs) { // 每一个patternsURI都代表一个文件
                        Path patternsPath = new Path(patternsURI.getPath());
                        String patternsFileName = patternsPath.getName().toString();
                        parseSkipFile(patternsFileName); // 将文件加入过滤范围，具体逻辑参见parseSkipFile(String
                                                            // fileName)
                    }
                }
            }
            /**
             * 将指定文件的内容加入过滤范围
             * 
             * @param fileName
             */
            private void parseSkipFile(String fileName) {
                try {
                    fis = new BufferedReader(new FileReader(fileName));
                    String pattern = null;
                    while ((pattern = fis.readLine()) != null) { // SkipFile的每一行都是一个需要过滤的pattern，例如\!
                        patternsToSkip.add(pattern);
                    }
                } catch (IOException ioe) {
                    System.err
                            .println("Caught exception while parsing the cached file '"
                                    + StringUtils.stringifyException(ioe));
                }
            }
            @Override
            public void map(Object key, Text value, Context context)
                    throws IOException, InterruptedException {
                // 这里的caseSensitive在setup()方法中赋值
                String line = (caseSensitive) ? value.toString() : value.toString()
                        .toLowerCase(); // 如果设置了大小写敏感，就保留原样，否则全转换成小写
                for (String pattern : patternsToSkip) { // 将数据中所有满足patternsToSkip的pattern都过滤掉
                    line = line.replaceAll(pattern, "");
                }
                StringTokenizer itr = new StringTokenizer(line); // 将line以\t\n\r\f为分隔符进行分隔
                while (itr.hasMoreTokens()) {
                    word.set(itr.nextToken());
                    context.write(word, one);
                    // getCounter(String groupName, String counterName)计数器
                    // 枚举类型的名称即为组的名称，枚举类型的字段就是计数器名称
                    Counter counter = context.getCounter(
                            MapCounters.class.getName(),
                            MapCounters.INPUT_WORDS.toString());
                    counter.increment(1);
                }
            }
        }
        /**
         * Reducer没什么特别的升级特性
         * 
         * @author Administrator
         */
        public static class IntSumReducer extends
                Reducer<Text, IntWritable, Text, IntWritable> {
            private IntWritable result = new IntWritable();
            public void reduce(Text key, Iterable<IntWritable> values,
                    Context context) throws IOException, InterruptedException {
                int sum = 0;
                for (IntWritable val : values) {
                    sum += val.get();
                }
                result.set(sum);
                context.write(key, result);
                Counter counter = context.getCounter(
                        ReduceCounters.class.getName(),
                        ReduceCounters.OUTPUT_WORDS.toString());
                counter.increment(1);
            }
        }
        public static void main(String[] args) throws Exception {
            Configuration conf = new Configuration();
            GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
            /**
             * 命令行语法是：hadoop command [genericOptions] [application-specific
             * arguments] getRemainingArgs()取到的只是[application-specific arguments]
             * 比如：$ bin/hadoop jar wc.jar WordCount2 -Dwordcount.case.sensitive=true
             * /user/joe/wordcount/input /user/joe/wordcount/output -skip
             * /user/joe/wordcount/patterns.txt
             * getRemainingArgs()取到的是/user/joe/wordcount/input
             * /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt
             */
            String[] remainingArgs = optionParser.getRemainingArgs();
            // remainingArgs.length == 2时，包括输入输出路径： 
            ///user/joe/wordcount/input /user/joe/wordcount/output
            // remainingArgs.length == 4时，包括输入输出路径和跳过文件：
            ///user/joe/wordcount/input /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt
            if (!(remainingArgs.length != 2 || remainingArgs.length != 4)) {
                System.err
                        .println("Usage: wordcount <in> <out> [-skip skipPatternFile]");
                System.exit(2);
            }
            Job job = Job.getInstance(conf, "word count");
            job.setJarByClass(WordCount2.class);
            job.setMapperClass(TokenizerMapper.class);
            job.setCombinerClass(IntSumReducer.class);
            job.setReducerClass(IntSumReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            List<String> otherArgs = new ArrayList<String>(); // 除了 -skip 以外的其它参数
            for (int i = 0; i < remainingArgs.length; ++i) {
                if ("-skip".equals(remainingArgs[i])) {
                    job.addCacheFile(new Path(remainingArgs[++i]).toUri()); // 将
                                                                            // -skip
                                                                            // 后面的参数，即skip模式文件的url，加入本地化缓存中
                    job.getConfiguration().setBoolean("wordcount.skip.patterns",
                            true); // 这里设置的wordcount.skip.patterns属性，在mapper中使用
                } else {
                    otherArgs.add(remainingArgs[i]); // 将除了 -skip
                                                        // 以外的其它参数加入otherArgs中
                }
            }
            FileInputFormat.addInputPath(job, new Path(otherArgs.get(0))); // otherArgs的第一个参数是输入路径
            FileOutputFormat.setOutputPath(job, new Path(otherArgs.get(1))); // otherArgs的第二个参数是输出路径
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        }
      
       
    来源：http://blog.csdn.net/u010967382/article/details/40152495