单词计数WordCountApp.class

public class WordCountApp {

    // 可以指定目录,目录下如果有二级目录的话,是不会执行的,只会执行一级目录.

    private static final String INPUT_PATH = "hdfs://hadoop1:9000/abd";// 输入路径

    private static final String OUT_PATH = "hdfs://hadoop1:9000/out";// 输出路径,reduce作业输出的结果是一个目录

    // _SUCCESS:在linux中,带下划线的这些文件一般都是被忽略不去处理的.表示作业执行成功.

    // _logs:产生的日志文件.

    // part-r-00000:产生的是我们的输出的文件.开始以part开始.r:reduce输出的结果,map输出的结果是m,00000是序号

    public static void main(String[] args) {

        Configuration conf = new Configuration();// 配置对象

        try {

            FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);

            fileSystem.delete(new Path(OUT_PATH), true);

            Job job = new Job(conf, WordCountApp.class.getSimpleName());// jobName:作业名称

            job.setJarByClass(WordCountApp.class);

            FileInputFormat.setInputPaths(job, INPUT_PATH);// 指定数据的输入

            job.setMapperClass(MyMapper.class);// 指定自定义map类

            job.setMapOutputKeyClass(Text.class);// 指定map输出key的类型

            job.setMapOutputValueClass(LongWritable.class);// 指定map输出value的类型

            job.setReducerClass(MyReducer.class);// 指定自定义Reduce类

            job.setOutputKeyClass(Text.class);// 设置Reduce输出key的类型

            job.setOutputValueClass(LongWritable.class);// 设置Reduce输出的value类型

            FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));// Reduce输出完之后,就会产生一个最终的输出,指定最终输出的位置

            job.waitForCompletion(true);// 提交给jobTracker并等待结束

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

    /**

     * 输入的key标示偏移量:这一行开始的字节. 输入的value:当前的行文本的内容. MapReduce执行过程:

     * 在这里边,我们的数据输入来自于原始文件,数据输出写出到hdfs, 中间的一堆都是map输出产生的临时结果.存放在map运行的linux磁盘上的,

     * 当经过shuffle时,reduce就会通过http把map端的对应数据给取过来.

     * mapred-default.xml中mapredcue.jobtracker

     * .root.dir,mapred.tmp.dir存储map产生的结果. 作业运行时产生这个目录,作业运行完之后它会删除目录.

     */

    public static class MyMapper extends

            Mapper<LongWritable, Text, Text, LongWritable> {

        // 源文件有两行记录,解析源文件会产生两个键值对.分别是<0,hello you>,<10,hello me>,所以map函数会被调用两次.

        // 在计算机存储的时候,是一维的结构.

        @Override

        protected void map(LongWritable key, Text value, Context context)

                throws IOException, InterruptedException {

            // 为什么要把hadoop类型转换为java类型?

            String line = value.toString();

            String[] splited = line.split("\t");

            // 使用hashMap写出去的优势:减少键值对出现的个数.

            Map<String, Integer> hashMap = new HashMap<String, Integer>();

            for (String word : splited) {

                // 在for循环体内,临时变量word出现的此时是常量1

                context.write(new Text(word), new LongWritable(1));// 把每个单词出现的次数1写出去.

            }

        }

    }

    // map函数执行结束后,map输出的<k,v>一共有4个.<hello,1>,<you,1>,<hello,1>,<me,1>

    // map把数据处理完之后,就会进入reduce.

    // 在进入shuffle之前,数据需要先进行分区.默认只有一个区.

    // 对每个不同分区中的数据进行排序,分组.

    // 排序后的结果:<hello,1>,<hello,1>,<me,1>,<you,1>

    // 分组后的结果(相同key的value放在一个集合中):<hello,{1,1}>,<me,{1}>,<you,{1}>

    // 规约(可选)

    // map中的数据分发到reduce的过程称作shuffle

    public static class MyReducer extends

            Reducer<Text, LongWritable, Text, LongWritable> {

        // 每一组调用一次reduce函数,一共调用了三次

        @Override

        protected void reduce(Text key, Iterable<LongWritable> values,

                Context context) throws IOException, InterruptedException {

            // count标示单词key在整个文件出现的次数

            // 分组的数量与reduce函数调用次数是相等的.

            // reduce函数调用次数与产生的<k,v>的数量抛开业务,没有任何关系!

            long count = 0L;

            for (LongWritable times : values) {

                count += times.get();

            }

            context.write(key, new LongWritable(count));

        }

    }

}