官网MapReduce实例代码详细批注

最后的WordCount v2.0,该代码相比源码中的org.apache.Hadoop.examples.WordCount要复杂和完整,更适合作为MapReduce模板代码

3.本文的目的就是为开发MapReduce的同学提供一个详细注释了的模板,可以基于该模板做开发。

--------------------------------------------------------------------------------

官网实例代码(略有改动)

WordCount2.java

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);
    }
}

Spark 颠覆 MapReduce 保持的排序记录 

内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:https://www.heiqu.com/b5cf427313beff8ef7c8efbd3edadca4.html