Hadoop-Mapreduce-英文单词计数（Brief版本-超详细解读）

package mapred;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.util.GenericOptionsParser;

// 统计文本中单词个数：客户端、作业驱动类
public class WordCountBrief {

    // 自定义Mapper静态内部Class 继承父类 Mapper类，进一步实现map过程
    // 这个Mapper静态内部类是一个泛型类型，它有4个形参类型(Object, Text, Text, IntWritable)，
    // 分别指定map函数的输入键、输入值、输出键、输出值的类型。
    // Hadoop 没有直接使用Java内嵌的类型，而是自己开发了一套可以优化网络序列化传输的基本类型。
    // 这些类型都在org.apache.hadoop.io包中，比如，此处的Object类型，适用于字段需要使用多种类型的情况，
    // Text类型相当于Java中的String类型，IntWritable类型相当于Java中的Integer类型
    public static class TokenizerMapper extends
            Mapper<Object, Text, Text, IntWritable> {

        // 定义2个变量
        // 定义 value变量，并赋值1， 每个单词计数1次，map的输出value即为1
        private final static IntWritable one = new IntWritable(1);
        // 定义 key变量
        private Text word = new Text();

        // 实现 map 函数逻辑
        // Context是Mapper的一个内部类，用于在Map或Reduce任务中跟踪Task的状态。
        // MapContext记录了Map执行的上下文，在Mapper类中，Context可以存储一些Job conf的信息，
        // 比如Job运行时的参数等，可以在map函数中处理这些信息。同时，Context也充当了Map和Reduce任务
        // 执行过程中各个函数的桥梁，这与Java Web中的session对象、application对象很相似。简单说，
        // Context对象保存了Job运行时的上下文信息，比如作业的配置信息、InputSplit信息、任务ID等，此处用到了Context的write方法
        // 此处map函数，有3个形参类型(Object key, Text value, Context context)，
        // 分别指定map函数的输入键、输入值、输出。
        public void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            // 构造一个用来解析输入value值的StringTokenizer对象
            StringTokenizer token = new StringTokenizer(value.toString());// Text类的value转化成字符串类型
            while (token.hasMoreTokens()) {// 循环取出token，并执行相关操作
                // 返回从当前位置到下一个分割符的字符串，给结果变量赋值
                word.set(token.nextToken());
                // map过程输出键值对：输出每个被拆分出来的单词，以及计数1； 进行输出
                context.write(word, one);
            }
        }
    }

    // 自定义Reducer Class 继承父类 Reducer类，进一步实现reduce过程
    // 这个Reducer静态内部类是一个泛型类型，它有4个形参类型(Text, IntWritable, Text, IntWritable)，
    // 分别指定reduce函数的输入键、输入值、输出键、输出值的类型。
    public static class IntSumReducer extends
            Reducer<Text, IntWritable, Text, IntWritable> {

        // 定义结果变量
        private IntWritable result = new IntWritable();

        // 实现 reduce 函数逻辑
        // 此处reduce函数，有3个形参类型(Text key, Iterable<IntWritable> values, Context context)，
        // 分别指定reduce函数的输入键、输入值列表或迭代器、输出。
        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);
        }
    }

    // 输入args为: hdfs://localhost:9000/words hdfs://localhost:9000/wordcountout{参数1为要统计的文件，参数2为指定的输出位置(必须是不存在的)}
    public static void main(String[] args) throws Exception {

        // Configuration类代表作业的配置。该类会加载core-site.xml、hdfs-site.xml、mapred-site.xml、hdfs-site.xml等配置文件
        // 任何作用的配置信息必须通过Configuration传递，通过Configuration可以实现在多个mapper和多个reducer任务之间共享信息
        Configuration conf = new Configuration();

        // 解析参数
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("Usage: wordcountbrief <in> [<in>...] <out>");
            System.exit(2);
        }

        // 删除已经存在的输出目录
        Path mypath = new Path(otherArgs[otherArgs.length - 1]);// 输出路径
        FileSystem hdfs = mypath.getFileSystem(conf);//获取hdfs文件系统
        // 如果文件系统中存在这个输出路径，则删除掉，保证输出目录不能提前存在
        if(hdfs.exists(mypath)){
            hdfs.delete(mypath, true);// true 代表递归删除
        }

        // 创建作业 job
        // Job对象指定了作业执行规范，可以用来控制整个作业的运行
        Job job = Job.getInstance(conf, "word count");

        // 通过传入的WordCountBrief类Class 设置job的jar包
        // 在Hadoop集群上运行的作业时，要把代码打包成一个jar文件，然后把此文件传到集群上，
        // 并通过命令来执行这个作业，但是命令中不必指定jar文件的名称。
        // 在这条命令中，通过Job对象的setJarByClass方法传递一个主类即可，
        // Hadoop会通过这个主类来查找包含它的jar包
        job.setJarByClass(WordCountBrief.class);

        //  设置Mapper类Class
        job.setMapperClass(TokenizerMapper.class);
        //  设置Combine类Class
        job.setCombinerClass(IntSumReducer.class);
        //  设置Reducer类Class
        job.setReducerClass(IntSumReducer.class);

        //  设置输出Key类Class
        job.setOutputKeyClass(Text.class);
        //  设置输出Value类Class
        job.setOutputValueClass(IntWritable.class);
        // 一般情况下，mapper和reducer输出的数据类型是一样的，所以可以用上面两条命令；
        // 如果不一样，可以用下面两条命令单独指定mapper输出的key、value数据类型
        // job.setMapOutputKeyClass(Text.class);
        // job.setMapOutputValueClass(IntWritable.class);

        // Hadoop 默认的是TextInputFormat和TextOutputFormat,此处可以显示地配置
        // job.setInputFormatClass(TextInputFormat.class);
        // job.setOutputFormatClass(TextOutputFormat.class);

        // 指定输入文件
        for (int i = 0; i < otherArgs.length - 1; ++i) {
            // FileInputFormat指定的路径可以是单个文件、一个目录或符合特定文件模式的一系列文件
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
        // 指定输出位置
        // 只能有一个输出路径，该路径指定的是reduce函数输出文件的写入目录。
        // 特别注意：输出目录不能提前存在，否则Hadoop会报错并拒绝执行作业，这样做的目的是防止数据丢失
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));

        // 提交job作业
        // waitForCompletion提交作业后，每秒会轮询作业进度，如果发现和上次报告后有改变，
        // 就把进度报告到控制台，作业完成后，如果成功就显示作业计数，如果失败则把导致作业失败的错误输出到控制台
        boolean result = job.waitForCompletion(true);

        System.exit(result ? 0 : 1);
    }
}
// 查看指定路径文件
// hadoop fs -ls hdfs://localhost:9000/
// 返回结果，如下：
// Found 4 items
//-rw-r--r--   3 jiangshan supergroup  573545760 2021-09-08 15:48 hdfs://localhost:9000/SBSNTEST111.txt
//-rw-r--r--   3 jiangshan supergroup         28 2021-09-08 16:01 hdfs://localhost:9000/testcreate
//drwxr-xr-x   - jiangshan supergroup          0 2021-09-08 20:22 hdfs://localhost:9000/wordcountout
//-rw-r--r--   3 jiangshan supergroup  163254798 2021-09-08 16:56 hdfs://localhost:9000/words

// 查看指定路径文件
//hadoop fs -ls hdfs://localhost:9000/wordcountout
//Found 2 items
//-rw-r--r--   3 jiangshan supergroup          0 2021-09-08 20:22 hdfs://localhost:9000/wordcountout/_SUCCESS
//-rw-r--r--   3 jiangshan supergroup      36850 2021-09-08 20:22 hdfs://localhost:9000/wordcountout/part-r-00000

// 将路径指定文件的内容输出到stdout
// hadoop fs -cat hdfs://localhost:9000/wordcountout/part-r-00000
// 返回结果，文本文件内容
// Case    2400
// China   4800
// Clip    2400
// Colony  2400
// Comparison      4800
// ComponentValue  2400
// ....................