走向云计算之MapReduce原理和运行流程详解

﹏ヽ暗。殇╰゛Y 2022-06-14 07:44 255阅读 0赞

## 一、概述 ##

MapReduce的设计理念源自于Google的MapReduce论文（发表于2004年12月），Hadoop MapReduce是Google MapReduce克隆版。MapReduce采用”分而治之”的思想，把对大规模数据集的操作，分发给一个主节点管理下的各个分节点共同完成，然后通过整合各个节点的中间结果，得到最终结果。简单地说，MapReduce就是”任务的分解与结果的汇总”。

MapReduce1.0的架构如下：  
![这里写图片描述][20170613202704568]  
在Hadoop1.x中，用于执行MapReduce任务的机器角色有两个：一个是JobTracker；另一个是TaskTracker，JobTracker是用于调度工作的，TaskTracker是用于执行工作的。一个Hadoop集群中只有一台JobTracker。

MapReduce2.0架构如下：  
![这里写图片描述][20170613202821980]  
MapReduce2.0架构是在YARN架构的基础上运行，具体可参考上一篇关于YARN架构的介绍博文。

在分布式计算中，MapReduce框架负责处理了并行编程中分布式存储、工作调度、负载均衡、容错均衡、容错处理以及网络通信等复杂问题，把处理过程高度抽象为两个函数：map和reduce，map负责把任务分解成多个任务，reduce负责把分解后多任务处理的结果汇总起来。需要注意的是，用MapReduce来处理的数据集（或任务）必须具备这样的特点：待处理的数据集可以分解成许多小的数据集，而且每一个小数据集都可以完全并行地进行处理。

## 二、MapReduce运行流程详解 ##

MapReduce将作业的整个运行过程分为两个阶段：Map阶段Reduce阶段。

Map阶段由一定数量的Map Task组成，例如：

*  输入数据格式解析：InputFormat
 *  输入数据处理：Mapper
 *  本地规约：Combiner（相当于local reducer，可选）
 *  数据分组：Partitioner

Reduce阶段由一定数量的Reduce Task组成，例如：

*  数据远程拷贝
 *  数据按照key排序
 *  数据处理：Reducer
 *  数据输出格式：OutputFormat

通常我们把从Mapper输出数据到Reduce读取数据之间的过程称之为shuffle。

一个MapReduce的基本流程图如下：  
![这里写图片描述][20170613204501952]  
现在来结合hadoop提供的WordCount程序来对其执行流程进行详解。源码如下：（针对本地环境对源码作了一些修改）

package com.kang;
    
    import java.io.IOException;
    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;
    
    public class WordCount {
    
        public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
    
            private final static IntWritable one = new IntWritable(1);
            private Text word = new Text();
    
            public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
                StringTokenizer itr = new StringTokenizer(value.toString());
                while (itr.hasMoreTokens()) {
                    word.set(itr.nextToken());
                    System.out.println("MapOut key:" + word.toString() + " value:" + one.toString());
                    context.write(word, one);
                }
            }
        }
    
        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) {
                    System.out.println("vlaue:" + val.toString());
                    sum += val.get();
                }
                result.set(sum);
                System.out.println("key:" + key.toString() + " valueSum:" + result.toString());
                context.write(key, result);
            }
        }
    
        public static void main(String[] args) throws Exception {
            Configuration conf = new Configuration();
            Job job = new Job(conf, "word count");
            job.setJarByClass(WordCount.class);
            job.setMapperClass(TokenizerMapper.class);
            job.setCombinerClass(IntSumReducer.class);
            job.setReducerClass(IntSumReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            FileInputFormat.addInputPath(job, new Path("hdfs://sparkproject1:9000/root/input/"));
            FileOutputFormat.setOutputPath(job, new Path("hdfs://sparkproject1:9000/root/output/"));
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        }
    }

### 1、Hadoop相关数据类型 ###

Hadoop提供了如下内容的数据类型，这些数据类型都实现了WritableComparable接口，以便用这些类型定义的数据可以被序列化进行网络传输和文件存储，以及进行大小比较。

BooleanWritable：标准布尔型数值
    ByteWritable：单字节数值
    DoubleWritable：双字节数
    FloatWritable：浮点数
    IntWritable：整型数
    LongWritable：长整型数
    Text：使用UTF8格式存储的文本
    NullWritable：当<key,value>中的key或value为空时使用

### 2、代码详解 ###

*  首先来看main函数

public static void main(String[] args) throws Exception {
            Configuration conf = new Configuration();
            Job job = new Job(conf, "word count");
            job.setJarByClass(WordCount.class);
            job.setMapperClass(TokenizerMapper.class);
            job.setCombinerClass(IntSumReducer.class);
            job.setReducerClass(IntSumReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            FileInputFormat.addInputPath(job, new Path("hdfs://sparkproject1:9000/root/input/"));
            FileOutputFormat.setOutputPath(job, new Path("hdfs://sparkproject1:9000/root/output/"));
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        }

在MapReduce中，由Job对象负责管理和运行一个计算任务，并通过Job的一些方法对任务的参数进行相关的设置。此处设置了使用TokenizerMapper完成Map过程中的处理；使用IntSumReducer完成Combine和Reduce过程中的处理。还设置了Map过程和Reduce过程的输出类型：key的类型为Text，value的类型为IntWritable。任务的输出和输入路径则由FileInputFormat和FileOutputFormat分别设定。完成相应任务的参数设定后，即可调用job.waitForCompletion()方法执行任务。

*  split阶段

![这里写图片描述][20170613211020548]  
在进行map计算之前，mapreduce会根据输入文件计算输入分片（input split），每个输入分片（input split）针对一个map任务，输入分片（input split）存储的并非数据本身，而是一个分片长度和一个记录数据位置的数组。输入分片（input split）往往和HDFS的block（块）关系很密切，假如我们设定HDFS的块的大小是64MB，如果我们输入有三个文件，大小分别是3MB、65MB和127MB，那么mapreduce会把3MB文件分为一个输入分片（input split），65MB则是两个输入分片（input split），而127MB也是两个输入分片（input split）。换句话说我们如果在map计算前不做输入分片调整（例如合并小文件），那么就会有5个map任务将执行，而且每个map执行的数据大小不均，这个也是mapreduce优化计算的一个关键点。

*  InputFormat（默认TextInputFormat）阶段

将文件拆分成splits后，MapReduce框架会根据用户指定的InputFormat格式进行数据分割，框架默认使用TextInputFormat格式。框架默认将文件按行分割形成`<key,value>`对，这一步由MapReduce框架自动完成，其中偏移量（即key值）包括了回车所占的字符数。  
![这里写图片描述][20170613212044108]

*  map阶段

map阶段就是程序员编写好的map函数了，因此map函数效率相对好控制，而且一般map操作都是本地化操作也就是在数据存储节点上进行。本例的map函数如下：

public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        
    
            private final static IntWritable one = new IntWritable(1);
            private Text word = new Text();
    
            public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
                StringTokenizer itr = new StringTokenizer(value.toString());
                while (itr.hasMoreTokens()) {
                    word.set(itr.nextToken());
                    System.out.println("MapOut key:" + word.toString() + " value:" + one.toString());
                    context.write(word, one);
                }
            }
        }

在MapReduce2.0中，Map过程需要继承org.apache.hadoop.mapreduce包中Mapper类，并重写其map方法。通过在map方法中添加两句把的key值和value值输出到控制台的代码，可以发现map方法中value值存储的是文本文件中的一行（以回车符为行结束标记），而key值为该行的首字母相对于文本文件的首地址的偏移量。然后StringTokenizer类将每一行拆分成为一个个的单词，并将`<word,1>`作为map方法的结果输出，其余的工作都交有MapReduce框架处理。  
这部分的工作可以用下图表示：  
![这里写图片描述][20170613212332503]

*  shuffle阶段

将map的输出作为reduce的输入的过程就是shuffle。它包含了map端的combine阶段、排序阶段、Partitioner阶段，reduce端的排序阶段等。下面进行一一详述。

1、combine阶段  
combiner阶段是程序员可以选择的，combiner其实也是一种reduce操作，因此我们看见WordCount类里是用reduce类来执行combine操作的。Combiner是一个本地化的reduce操作，它是map运算的后续操作，主要是在map计算出中间文件前做一个简单的合并重复key值的操作。例如我们对文件里的单词频率做统计，map计算时候如果碰到一个hadoop的单词就会记录为1，但是这个hadoop单词可能会出现n多次，那么map输出文件冗余就会很多。因此在reduce计算前先对相同的key做一个合并操作，那么文件会变少，这样就提高了宽带的传输效率。毕竟hadoop计算力宽带资源往往是计算的瓶颈也是最为宝贵的资源。  
但是combiner操作是有风险的，使用它的原则是combiner的输入不会影响到reduce计算的最终输入。例如：如果计算只是求总数，最大值，最小值可以使用combiner。但是做平均值计算使用combiner的话，最终的reduce计算结果就会出错。  
对于Hadoop而言，得到map方法输出的`<key,value>`对后，Mapper会将它们按照key值进行排序，并执行Combine过程，将key至相同value值累加，得到Mapper的最终输出结果。当然我们可以把这个排序归入到combine过程中。  
该阶段如下图所示：（这里使用reduce函数来进行combine操作，见上面源码）  
![这里写图片描述][20170613213559436]

2、Partitioner阶段  
Shuffle一开始就是map阶段做输出操作，一般mapreduce计算的都是海量数据，map输出时候不可能把所有文件都放到内存操作，中间数据是存储在磁盘上的。map写入磁盘的过程十分的复杂，更何况map输出时候要对结果进行排序，内存开销是很大的。map在做输出时候会在内存里开启一个环形内存缓冲区，这个缓冲区专门用来输出的，同时map还会为输出操作启动一个守护线程，如果缓冲区的内存达到了阀值的80%时候，这个守护线程就会把内容写到磁盘上，这个过程叫spill。等map输出全部做完后，map会合并这些输出文件。这个过程里有一个Partitioner操作，Partitioner操作和map阶段的输入分片（Input split）很像。一个Partitioner对应一个reduce作业，如果我们mapreduce操作只有一个reduce，那么Partitioner就只有一个，如果我们有多个reduce，那么Partitioner对应的就会有多个。Partitioner因此就是reduce的输入分片，这个程序员可以编程控制，主要是根据实际key和value的值，根据实际业务类型或者为了更好的reduce负载均衡要求进行，这是提高reduce效率的一个关键所在。到了reduce阶段就是合并map输出文件了，Partitioner会找到对应的map输出文件，然后进行复制操作，复制操作时reduce会开启几个复制线程，这些线程默认个数是5个，程序员也可以在配置文件更改复制线程的个数。这个复制过程和map写入磁盘过程类似，也有阀值和内存大小，阀值一样可以在配置文件里配置，而内存大小是直接使用reduce的tasktracker的内存大小，复制时候reduce还会进行排序操作和合并文件操作，这些操作完了就会进行reduce计算了。  
Partitioner默认实现是采用哈希函数：hash(key) mod R，其中  
R是Reduce Task数目，该过程允许用户自定义。  
对于本程序而言，只有一个Reduce Task，所以两个map的输出会归并到该reduce下，如下图所示：  
![这里写图片描述][20170613215019071]  
  
从图中可以看出，正真reduce方法接收到数据时，value的形式是`Iterable<IntWritable> values`的形式，从上述源码中也可以看出来。

*  reduce阶段

Reduce方法需要继承org.apache.hadoop.mapreduce包中Reducer类，并重写其reduce方法。Map过程输出`<key,values>`中key为单个单词，而values是对应单词的计数值所组成的列表。Map的输出就是Reduce的输入，所以reduce方法只要遍历values并求和，即可得到某个单词的总次数。其源码如下：

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;
                System.out.println("ReduceIn key:"+key.toString()+" value:"+values.toString());
                for (IntWritable val : values) {
                    System.out.println("vlaue:" + val.toString());
                    sum += val.get();
                }
                result.set(sum);
                System.out.println("ReduceOut key:" + key.toString() + " valueSum:" + result.toString());
                context.write(key, result);
            }
        }

reduce过程如下图：  
![这里写图片描述][20170613215958980]

### 3、从整体上来看MapReduce ###

这里给出MapReduce1.0的整体处理流程，对于MapReduce2.0，只是把jobTracker的任务分为两部分，其他逻辑基本一致。参考YARN的运行流程即可。  
![这里写图片描述][20170614103546162]

## 三、程序提交集群测试 ##

在本地eclipse进行开发完成后，可以把程序打成jar包，放到集群上去运行。基本过程如下：  
1、项目打包

右键项目，选择“export”选项，选择java下的jar选项。如图：  
![这里写图片描述][20170614110932061]  
点击下一步，选择jar包存放位置和名字，其他默认即可。点击finish完成。

2、将jar包上传至linux集群，此过程略。

3、进入jar包所在目录，在该目录输入如下命令启动任务：

hadoop jar wordcount.jar com.kang.WordCount /root/input /root/output

注意：我的linux上已经把hadoop的bin目录加入了环境变量的Path中，所以可以在任何目录下直接运行hadoop命令。  
命令中的`wordcount.jar`是我们打包上传的jar包名称，`com.kang.WordCount`是main方法所在的类的全类名（包含包名），`/root/input`是输入目录，`/root/output`是输出目录。需要注意的是输出目录不能在HDFS上已存在，否则hadoop运行时会报错。  
事实上对于本次实例来说，不必指定输入输出目录，因为我在源程序中已经把目录指定了（见上面源码）。如果我们查看hadoop的wordcount原版源码，可以看到其main方法如下：

public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
          System.err.println("Usage: wordcount <in> [<in>...] <out>");
          System.exit(2);
        }
        Job job = new Job(conf, "word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        for (int i = 0; i < otherArgs.length - 1; ++i) {
        
          FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
        FileOutputFormat.setOutputPath(job,
          new Path(otherArgs[otherArgs.length - 1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
      }

可以看到它是通过命令行参数指定的输入输出目录。实际开发中我们可以灵活使用。

运行结果如下图：  
![这里写图片描述][20170614112056981]  
![这里写图片描述][20170614112111265]  
![这里写图片描述][20170614112128781]  
![这里写图片描述][20170614112148547]

4、查看结果：  
通过命令：`hadoop fs -ls /root/output`可查看输出目录文件列表。  
通过命令：`hadoop fs -cat /root/output/part-r-00000`查看结果。  
![这里写图片描述][20170614112409332]

## 四、常见MapReduce应用场景 ##

*  简单的数据统计，比如网站pv、uv统计
 *  搜索引擎建索引
 *  海量数据查找
 *  复杂数据分析算法实现，例如  
    聚类算法  
    分类算法  
    推荐算法  
    图算法

[20170613202704568]: /images/20220614/260c76c45f414d1dafd0ca312a12ad5d.png
[20170613202821980]: /images/20220614/a97855125cea46e8a8eb230f27580560.png
[20170613204501952]: /images/20220614/0b30f4ac9e81489a8b732804e7c2d2c4.png
[20170613211020548]: /images/20220614/7b3a9411299f408290b8dc3bb65f4c08.png
[20170613212044108]: /images/20220614/ecbb96ca482045d690b0f81cd3343161.png
[20170613212332503]: /images/20220614/94665e46c65343c5972a458add4ab056.png
[20170613213559436]: /images/20220614/9849ebbec9654f579d936257c93f0d18.png
[20170613215019071]: /images/20220614/5758f1441e7047d9b4257e6962cede1b.png
[20170613215958980]: /images/20220614/dc4d98c711b94f99868c83bf855e7302.png
[20170614103546162]: /images/20220614/0da090be782a4be9a54364d6e3da1fcc.png
[20170614110932061]: /images/20220614/aa98e7e5ad6840cc919274dc7c8e1a1d.png
[20170614112056981]: /images/20220614/3611094af3324b1d8024aa8242bd4c0b.png
[20170614112111265]: /images/20220614/b959efe1dda948529e7e5a4c16e89dda.png
[20170614112128781]: /images/20220614/d77687c186db4c63a6705467ff8fe966.png
[20170614112148547]: /images/20220614/8f75d097011b40019c343f37beed18ca.png
[20170614112409332]: /images/20220614/d52b4fedd99c44f7bb360c2d02afcfba.png