Hadoop大数据开发基础系列：四、MapReduce初级编程

2.通过源码初识MapReduce工程

此部分，目的是对MapReduce的核心模块 Mapper与Reducer的执行流程有一定的认识。通过学习wordcount的源码来了解一下。

    2.1 通俗理解MapReduce原理

        (1)MapReduce包括Mapper模块和Reducer模块，MapReduce可以看做是一个专业处理大数据的工程队，主要由下面成员构成：

            ①Mapper：映射器 ②Mapper助理InputFormat：输入文件读取器

            ③Shuffle：运输队 ④Shuffle助理Sorter：排序器 

            ⑤Reducer：归约器 ⑥Reducer助理OutputFormat：输出结果写入器

        (2)简化的MapReduce处理流程图：

            ①数据切片：系统把数据分给多个Mapper来处理，通过数据分片的形式，这是分布式计算的第一步。

            ②数据映射：分片完成后，Mapper助理InputFormat从文件输入目录读取数据，再由Mapper对数据进行解析，组织成为新的格式（键值对形式），最后Mapper将处理好的数据输出，等待shuffle运输队取走结果。

            ③数据混洗：shuffle运输队把获取的结果按照相同的键(Key)进行汇集，再把结果送到Shuffle助理Sorter处，由Sorter负责对这些结果排序，，然后提交给Reducer。

            ④数据归约：Reducer收到数据后，将结果进行汇总与映射工作，得到最终的计算结果，最后由Reducer助理OutputFormat将结果输出到指定位置处。

    2.2 了解MR实现词频统计的执行流程

        下面举一个实例来说明一下Map和Reduce过程

        (1)Map任务的处理过程

        (2)Reduce任务的处理过程

    2.3 读懂官方提供的WordCount源码

        要编写数据处理程序，还要参考MapReduce编程的具体规范，下面进行代码级别的分析和说明：

        在“D:\tools\hadoop-2.7.7\share\hadoop\mapreduce\sources”目录下找到“hadoop-mapreduce-examples-2.7.7-sources.jar”，解压缩此文件，在子目录“\org\apache\hadoop\examples”看到WordCount文件，这就是WordCount程序的源代码。

        从结构上可以分为3部分，分别是应用程序Driver、Mapper模块与Reducer模块。

        (1)应用程序Driver分析

            这里的Driver程序主要是指的main函数，在main函数里面进行MapReduce程序的一些初始化设置，并提交任务，等待程序运行完成。

            基本上是这样的格式，在此基础上只需要修改部分参数即可。

        (2)Mapper模式分析

        (3)Reducer模式分析

        (4)概括地讲：

            进行MapReduce编程时，开发者主要处理的是Mapper和Reducer两个模块，其中包括定义输入输出的键值对格式、编写map与reduce函数中定义的处理逻辑等。

3.编程实现按日期统计访问次数

    本部分任务目标是统计用户在2016年每个自然日的总访问次数。原始数据文件中提供了用户名称与访问日期，这个任务实质就是要获取以每个自然日为单位的所有用户访问次数的累加值。如果通过MapReduce编程实现这个任务，首先要考虑的是，Mapper与Reducer各自的处理逻辑是怎样的，然后根据处理逻辑编写核心代码，最后在Eclipse中编写核心代码，编译打包后提交集群运行。

    3.1 分析思路与处理逻辑

        着重考虑以下几个要素：

        ①输入输出格式 ②Mapper要实现的逻辑 ③Reducer要实现的计算逻辑

        (1) 定义输入/输出格式 

            社交网站用户的访问日期在格式上属于文本格式，访问次数为整型数值格式。其组成的键值对为<访问日期，访问次数>，因此Mapper的输出与Reducer的输出都选用Text类与IntWritable类。

        (2)Mapper 类的逻辑实现

            Mapper类中最主要的部分就是map函数。map函数的主要任务就是读取用户访问文件中的数据，输出所有访问日期与初始次数的键值对。因此访问日期是数据文件的第二列，所有先定义一个数组，再提取第二个元素，与初始次数1一起构成要输出的键值对，即<访问日期，1>。

        (3)Reducer的逻辑实现

            Reducer类中最主要的部分就是reduce函数。reduce的主要任务就是读取Mapper输出的键值对<访问日期，1>。这一部分与官网给出的WordCount中的Reducer完全相同。

    3.2 编写核心模块代码

        目录结构：

package test;
import java.io.IOException;
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;
public class DailyAccessCount {
       // Mapper模块
       public static class MyMapper
    extends Mapper<Object, Text, Text, IntWritable>{
             private final static IntWritable one = new IntWritable(1);
             public void map(Object key, Text value, Context context) //map函数的编写要根据读取的文件内容和业务逻辑来写
                           throws IOException, InterruptedException {
                    String line = value.toString();
                    String array[] = line.split(",");//指定,为分隔符，组成数组
                    String keyOutput = array[1];//提取数组中的访问日期作为Key
                    context.write(new Text(keyOutput), one);//形成键值对
             }
       }
       
       // Reducer模块
       public static class MyReducer
    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;                                   //定义累加器，初始值为0
                    for (IntWritable val : values) {
                           sum += val.get();                       //将相同键的所有值进行累加
                    }
                    result.set(sum);
                    context.write(key, result);
             }
       }
       
       //Driver模块，主要是配置参数
       public static void main(String[] args) throws Exception {
           Configuration conf = new Configuration();
           
           Job job = Job.getInstance(conf, "DailyAccessCount");
           job.setJarByClass(DailyAccessCount.class);
           job.setMapperClass(MyMapper.class);
           job.setReducerClass(MyReducer.class);
           job.setMapOutputKeyClass(Text.class);
           job.setMapOutputValueClass(IntWritable.class);
           job.setOutputKeyClass(Text.class);
           job.setOutputValueClass(IntWritable.class);
           for (int i = 0; i < args.length - 1; ++i) {
             FileInputFormat.addInputPath(job, new Path(args[i]));
           }
           FileOutputFormat.setOutputPath(job,
             new Path(args[args.length - 1]));
           System.exit(job.waitForCompletion(true) ? 0 : 1);
         }    
}

    3.3 任务实现

        将文件编译生成JAR包文件，提交Hadoop集群执行

    在运行过程中报错，原因是我在外面的JDK用的是1.9的，等级过高了（Linux系统的JDK是1.8的），所以要重新配置JDK。

    记住，在用Windows环境下的JDK要和Hadoop集群环境下的JDK环境相同。

    历史各个版本的JDK下载地址： 

                      https://www.oracle.com/technetwork/java/javase/downloads/java-archive-javase8-2177648.html

    经过配置，终于可以运行啦，哈哈哈 o(*￣︶￣*)o

hadoop jar NewDaily.jar test.NewDaily /user/dftest/user_login.txt /user/dftest/AccessCount

    结果如下：

        再来查看输出结果：

        打开文件可以看到：

        第一列是已经按照自然日期排好顺序，第二列是对应日期的总访问次数，任务基本完成。

4.编程实现按访问次数排序

    前一部分完成了日期统计任务，本部分要对AccessCount中的数据按照访问次数进行排序，将排序后的结果存放在相同目录下的TimesSort中。

    4.1 分析思路与处理逻辑

        MapReduce只会对键值进行排序，所以我们在Mapper模块中对于输入的键值对，把Key与Value位置互换，在Mapper输出后，键值对经过shuffle的处理，已经变成了按照访问次数排序的数据顺序啦，输出格式为<访问次数，日期>。Reducer的处理和Mapper恰好相反，将键和值的位置互换，输出格式变为<日期，访问次数>。

    4.2 编写核心模块代码

package test;
import java.io.IOException;
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;
public class AccessTimesSort {
       // Mapper模块
       public static class MyMapper
    extends Mapper<Object, Text, IntWritable,Text>{
             public void map(Object key, Text value, Context context) //map函数的编写要根据读取的文件内容和业务逻辑来写
                           throws IOException, InterruptedException {
                    String line = value.toString();
                    String array[] = line.split("\t");//指定,为分隔符，组成数组
                    int keyOutput = Integer.parseInt(array[1]);//提取数组中的访问次数作为Key
                    String valueOutput = array[0]; //将日期作为value
                    context.write(new IntWritable(keyOutput), new  Text(valueOutput));
             }
       }
       
       // Reducer模块
       public static class MyReducer
    extends Reducer<IntWritable,Text,Text,IntWritable> {//注意与上面输出对应
             public void reduce(IntWritable key, Iterable<Text> values, Context  context)  
                           throws IOException, InterruptedException {
                    for (Text val : values) {
                           context.write(val, key);                //进行键值位置互换
                    }
             }
       }
       
       //Driver模块，主要是配置参数
       public static void main(String[] args) throws Exception {
           Configuration conf = new Configuration();
           
           Job job = Job.getInstance(conf, "AccessTimesSort");
           job.setJarByClass(AccessTimesSort.class);
           job.setMapperClass(MyMapper.class);
           job.setReducerClass(MyReducer.class);
           job.setMapOutputKeyClass(IntWritable.class);
           job.setMapOutputValueClass(Text.class);
           job.setOutputKeyClass(Text.class);
           job.setOutputValueClass(IntWritable.class);
           for (int i = 0; i < args.length - 1; ++i) {
             FileInputFormat.addInputPath(job, new Path(args[i]));
           }
           FileOutputFormat.setOutputPath(job,
             new Path(args[args.length - 1]));
           System.exit(job.waitForCompletion(true) ? 0 : 1);
         }    
}

    4.3 任务实现

        (1)编译生成jar包

        (2)将jar包上传到集群

        (3)执行jar包内的AccessTimesSort类

hadoop jar NewDaily.jar test.AccessTimesSort /user/dftest/AccessCount /user/dftest/TimesSort

        (4)查看执行结果(由此可以看到结果为升序排列)

        此任务顺利完成 哈哈。

5.小结

    本章介绍了MapReduce编程的基础知识，通过对Hadoop官方的示例代码的分析及解读，深入了解了MapReduce的执行过程。MapReduce把复杂的、运行在Hadoop集群上的并行计算过程集成到了两个模块——Mapper和Reducer上。开发人员只需要把业务处理逻辑通过其中的map函数和reduce函数来实现，就可以达到分布式并行编程的目的。

    MapReduce执行过程主要包括以下几个部分：读取分布式文件系统的数据，进行数据分片，执行map任务以输出中间结果，shuffle阶段把中间结果进行汇合、排序，再传到Reduce任务，在Reduce阶段对数据进行处理，输出最终结果到分布式文件系统内。

6.实训

    实训目的是，掌握MapReduce编程的基本方法，通过MapReduce编程来实现一些常用的数据处理方法，包括求最大值、去重等。

    实训1.获取成绩表的最高分记录

        (1)需求说明：对于样例文件subject_score，即成绩表A。文件中的每一行数据包含两个字段：科目和分数。要求获得成绩列表中每个科目成绩最高的记录，并将结果输出到最高成绩表B。

        表A的部分内容：

        要输出的表B结构：

        (2)实现思路与步骤：

            ①在Mapper中，map函数读取成绩表A中的数据，直接将读取的数据以空格分隔，组成键值对<科目，成绩>，即设置输出键值对类型为<Text，IntWritable>。

            ②在Reducer中，由于map函数输出键值对类型是<Text，IntWritable>，所以在Reducer中接收的键值对类型就是<Text，Iterable<IntWritable>>。针对相同的键遍历它的值，找到最高值，最后输出的键值对为<科目，最高成绩>。

        (3)实现及输出结果：

            ①代码实现：

package test;


import java.io.IOException;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
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;


public class ScoreSorting {
    // Mapper模块
        public static class MyMapper
        extends Mapper<LongWritable, Text, Text ,IntWritable>{
            Text course=new Text();
            IntWritable score=new IntWritable();
            public void map(LongWritable key, Text value,
                    Mapper<LongWritable, Text, Text ,IntWritable>.Context context) //map函数的编写要根据读取的文件内容和业务逻辑来写
                    throws IOException, InterruptedException {
                String line = value.toString();
                String array[] = line.trim().split(" ");//trim函数去掉两边多余的空格，指定空格为分隔符，组成数组
                course.set(array[0]);//第一列是科目
                score.set(Integer.parseInt(array[1]));//第二列是分数
                context.write(course, score);
            }
        }
        
        // Reducer模块
        public static class MyReducer
        extends Reducer<Text,IntWritable,Text,IntWritable> {//注意与上面输出对应
            private IntWritable result = new IntWritable();
            public void reduce(Text key, Iterable<IntWritable> values,
                    Reducer<Text,IntWritable,Text,IntWritable>.Context context)  
                    throws IOException, InterruptedException {
                int maxscore=0;                        //初始化最大值
                for (IntWritable score:values) {
                    if(maxscore < score.get()) {
                        maxscore=score.get();        //相同键内找最大值
                    }
                }
                result.set(maxscore);
                context.write(key, result);
                
            }
        }
        
        //Driver模块，主要是配置参数
        public static void main(String[] args) throws Exception {  //对有几个参数要有很强的敏感性，如果多可以用前面的遍历方式，如果少就可以直接指定。
            if(args.length!=2) {
                System.err.println("ScoreSorting <input> <output>");
                System.exit(-1);
            }
            Configuration conf = new Configuration();  
            Job job = Job.getInstance(conf, "ScoreSorting");
            job.setJarByClass(ScoreSorting.class);
            job.setMapperClass(MyMapper.class);
            job.setReducerClass(MyReducer.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            job.setNumReduceTasks(1);
            FileInputFormat.addInputPath(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job,new Path(args[1]));
            System.exit(job.waitForCompletion(true) ? 0 : 1);
          }    
}

            上传数据文件并执行程序：

hdfs dfs -put /testhadoop/subject_score.txt /user/dftest
hadoop jar NewDaily3.jar test.ScoreSorting /user/dftest/subject_score.txt /user/dftest/SortScore8     //经过8次才调好的。

经过不断地调试，但是总会出现输入类型不匹配的问题，最终，找到在map函数重写的时候，因为值类型错了，应该是Text类型，写成了IntWritable类型，但是报错总是报是因为Text 和 LongWritable问题，所以有点迷。

最后问题解决：

    实训2.对两个文件中的数据进行合并和去重

        (1)需求说明：

            有两个样例文件：XX与YY。要求合并两个文件中的数据，并对合并后的数据进行去重，将结果输出到文件ZZ。

        (2)实现思路与步骤：

            ①利用MapReduce中Reducer类会合并相同键值对的特性，对目标数据进行去重。

            ②在HDFS创建目录XXYY，将样例文件XX与YY上传到此目录。MapReduce程序读取此目录下的文件。

            ③在Mapper类中，map函数读取两个文件数据，直接将读取的数据作为键，将值设置为1，最后输出格式为<Text,IntWritable>。

            ④在Reducer中，键保持不变，将对应的值取为空，输出类型为<Text，NullWritable>。

        (3)实现及输出结果：

            略

7.小练习

    在MapReduce程序中，Reducer类中包括的函数有：B

    A.   startup、reduce、end        B.    setup、reduce、cleanup    

    C.    start、run、reduce、end    D.   startup、run、end     

下一章将对MapReduce编程进行更深一步的剖析 ^_^ 。