Hadoop教程第五弹 MapReduce工作原理

╰半夏微凉° 2023-01-22 14:58 61阅读 0赞

# 1、MapReduce概述 #

MapReduce是Hadoop系统中最重要的计算引擎，它不仅直接支持交互式应用、基于程序的应用，而且还是Hive等组件的基础。MapReduce v2(也就是Yarn)则进一步提升了该计算引擎的性能和通用性。  
在Hadoop平台上，MapReduce框架负责处理并行编程中分布式存储、工作调度、负载均衡、容错及网络通信等复杂工作，把处理过程高度抽象为两个函数：Map和Reduce。Map负责把作业分解成多个任务，Reduce负责把分解后多任务处理的结果汇总起来。

在MapReduce运行过程中有三类进程：MrAppMaster、MapTask和ReduceTask。MrAppMaster负责整个程序的过程调度及状态协调。MapTask阶段并发运行，ReduceTask阶段也是并发运行（当有多个ReduceTask时，比如一个ReduceTask负责统计a-n开头的单词，另一个统计o-z开头的单词）。

在Hadoop中，用于执行MapReduce作业的机器角色有两个：JobTracker和TaskTracker。JobTracker用于调度作业，TaskTracker用于跟踪任务的执行情况。一个Hadoop集群只有一个JobTracker。

需要注意的是，用MapReduce来处理的数据集必须具备这样的特点：数据集可以分解成许多小的数据集，而且每一个小数据集都可以完全独立地并行处理。

在MapReduce程序中计算的数据可以来自多个数据源，如本地文件、HDFS、数据库等。最常用的是HDFS,同时，在计算完成后也可以将结果存储到HDFS。  
当MapReduce运行Task时会基于用户编写的业务逻辑进行读取或存储数据。

Yarn的基本思想是将JobTracker的资源管理和作业的调度、监控两大主要职能拆分为两个独立的进程：一个全局的ResourceManager和与每个应用对应的Application Master。

MapReduce优点：

*  易于编程：简单地实现一些接口，就可以完成一个分布式程序；
 *  良好的扩展性：当计算资源不能满足时，可以方便的增加机器来扩展计算能力；
 *  高容错性：一台机器挂了，MapReduce可以自动地将计算任务转移到另一个节点；
 *  适合PB级以上海量数据的离线处理；

缺点：

*  不擅长实时计算；
 *  不擅长流式计算；
 *  不擅长迭代式计算（如果要强制这么做，那必须有多个MapReduce任务，且中间结果会写入到磁盘，效率低）；

# 2、Hadoop序列化机制 #

Hadoop序列化特点：

*  紧凑：节约存储空间；
 *  快速：读写数据的额外开销小；
 *  互操作：支持多语言交互；

常见的Java类型与Hadoop Writable类型对照表如下所示。

<table style="width:500px;"> 
 <thead> 
  <tr> 
   <th>Java 类型</th> 
   <th>Hadoop Writable 类型</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Boolean</td> 
   <td>BooleanWritable</td> 
  </tr> 
  <tr> 
   <td>Byte</td> 
   <td>ByteWritable</td> 
  </tr> 
  <tr> 
   <td>Int</td> 
   <td>IntWritable</td> 
  </tr> 
  <tr> 
   <td>Float</td> 
   <td>FloatWritable</td> 
  </tr> 
  <tr> 
   <td>Long</td> 
   <td>LongWritable</td> 
  </tr> 
  <tr> 
   <td>Double</td> 
   <td>DoubleWritable</td> 
  </tr> 
  <tr> 
   <td>String</td> 
   <td>Text</td> 
  </tr> 
  <tr> 
   <td>Map</td> 
   <td>MapWritable</td> 
  </tr> 
  <tr> 
   <td>Array</td> 
   <td>ArrayWritable</td> 
  </tr> 
  <tr> 
   <td>Null</td> 
   <td>NullWritable</td> 
  </tr> 
 </tbody> 
</table>

也可以自定义序列化类型。自定义序列化类型注意事项 ：

*  实现Writable接口，重写序列化方法和反序列化方法，且序列化顺序与反序列化顺序保持一致；
 *  定义无参构造方法，反序列化反射时会用到；
 *  如果自定义bean在key中传输，需要重写Comparable接口，因为MapReduce中的Shuffle过程要求key必须能排序；

自定义序列化类型FlowBean的代码如下所示。

import org.apache.hadoop.io.Writable;
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    public class FlowBean implements Writable {
        /**
         * 上行流量
         */
        private long upFlow;
        /**
         * 总流量
         */
        private long downFlow;
        /**
         * 下行流量
         */
        private long totalFlow;
    
        public long getUpFlow() {
            return upFlow;
        }
    
        public void setUpFlow(long upFlow) {
            this.upFlow = upFlow;
        }
    
        public long getDownFlow() {
            return downFlow;
        }
    
        public void setDownFlow(long downFlow) {
            this.downFlow = downFlow;
        }
    
        public long getTotalFlow() {
            return totalFlow;
        }
    
        public void setTotalFlow(long totalFlow) {
            this.totalFlow = totalFlow;
        }
    
        @Override
        public void write(DataOutput dataOutput) throws IOException {
            dataOutput.writeLong(upFlow);
            dataOutput.writeLong(downFlow);
            dataOutput.writeLong(totalFlow);
        }
    
        @Override
        public void readFields(DataInput dataInput) throws IOException {
            this.upFlow = dataInput.readLong();
            this.downFlow = dataInput.readLong();
            this.totalFlow = dataInput.readLong();
        }
    
        @Override
        public String toString() {
            return "upFlow="+upFlow +",downFlow="+downFlow+",totalFlow="+totalFlow;
        }
    }

# 3、手写WordCount #

编写一个MapReduce程序需要分三步：

*  编写Mapper：用户编写的map业务逻辑。需要继承Mapper类，并重写map方法；Mapper类的泛型分别为KEYIN、KEYOUT、VALUEIN、VALUEOUT。
 *  编写Reducer：用户编写的reduce业务逻辑。需要继承Reducer类，并重写reduce方法；Reducer类的泛型分别为KEYIN、KEYOUT、VALUEIN、VALUEOUT。
 *  编写Driver：相当于yarn集群的客户端，用于提交任务到yarn中运行；

在Mapper阶段，默认情况下，Mapper的输入数据的key是该行数据的偏移量，value是该行数据。对每一行数据调用一次map方法。

在Reducer阶段，Reducer的输入正是Mapper的输出，且对每一组具有相同key的kv对(Mapper的输出)调用一次reduce方法；

## 3.1、编写Mapper ##

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    /**
     * KEYIN 偏移量，long类型
     * VALUEIN 一行数据，String类型
     * KEYOUT 一个单词，String类型
     * VALUEOUT 数量，int类型
     */
    public class WordMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
        private IntWritable count = new IntWritable(1);
        private Text keyOut = new Text();
        /**
         * 每行数据调用一次map方法
         */
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] wordArr = line.split(" ");
            for (String word : wordArr) {
                keyOut.set(word);
                context.write(keyOut,count);
            }
    
        }
    }

## 3.2、编写Reducer ##

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    public class WordReducer extends Reducer<Text, IntWritable,Text,IntWritable> {
        private IntWritable valueOut = new IntWritable();
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int count = 0;
            for (IntWritable value : values) {
                count += value.get();
            }
            valueOut.set(count);
            context.write(key,valueOut);
        }
    }

## 3.3、编写Driver ##

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.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    
    public class WordCountDriver {
    
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            // 1、获取配置信息及job对象
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
            // 2、关联本Driver程序的jar
            job.setJarByClass(WordCountDriver.class);
            // 3、关联Mapper和Reducer
            job.setMapperClass(WordMapper.class);
            job.setReducerClass(WordReducer.class);
            // 4、设置Mapper输出类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
            // 5、设置最终输出类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            // 6、设置输入和输出路径
            FileInputFormat.setInputPaths(job,new Path(args[0]));
            FileOutputFormat.setOutputPath(job,new Path(args[1]));
            // 7、提交job
            boolean b = job.waitForCompletion(true);
            System.exit(b?0:1);
        }
    
    }

## 3.4、运行 ##

最后，将wordcount项目打成jar包，上传到Linux机器上，并执行如下命令运行wordcount。

hadoop jar wordcount.jar /input /output

# 4、MapReduce框架原理 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70][]

## 4.1、InputFormat ##

首先介绍一下切片这个概念。

HDFS是以块存储的，块对文件的切割是物理上的，而数据切片是逻辑上的，切片只会记录索引，不会真正地切分文件。

我们知道，MrAppMaster会启动多个MapTask来处理任务，那到底启动多少个MapTask(MapTask并行度)呢？

切片数=MapTask数

那切片数是如何决定的呢？先来看这样一个场景。

1G的数据，启动多个MapTask，可提高处理速度。但1kB的数据，如果启动多个MapTask，可能启动时间都超过处理时间了，就没必要启动多个MapTask。

InputFormat有一个子类是FileInputFormat，FileInputFormat的数据源是文件，TextInputFormat(按行读取)是默认的FileInputFormat实现类。  
切片大小计算公式：computeSplitSize(Math.max(minSize,Math.min(maxSize,blocksize)))  
默认的mapreduce.input.fileinputformat.split.minsize=1  
默认的mapreduce.input.fileinputformat.split.maxsize=Long

因此，默认的切片大小=blocksize=128M  
另外，如果一个文件是128.1M，理论上要切成两片，但这似乎不太合理，因此是否切片还要看是否超过1.1倍,很明显128.1要小于128的1.1倍，那么就不会被切成两片。

切片不考虑整体数据集，而是对单个文件进行切片。

在提交Job时，任务分两种形式运行：本地模式和集群模式，本地模式的块大小是32M。

由于默认的TextInputFormat不管文件多小都会占一个切片，当有大量小文件时会严重消耗内存，可用专门用于处理大量小文件的CombineTextInputFormat代替。

## 4.2、MapReduce工作流程 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70 2][]

上面两张图就是MapReduce的工作流程，下面来用文字叙述一些要注意的点。

a）当客户端将job提交到Yarn集群时，Yarn会开启MrAppMaster，MrAppMaster会读取客户端的提交信息（切片信息、jar、job.xml，如果是本地模式运行则不需要jar，job.xml封装了任务运行时的各种信息）。MrAppMaster读取切片信息并开启对应个数的MapTask。

b）MapTask的处理结果会输送到outputCollector，outputCollector就是环形缓冲区。下图也是一个环形缓冲区。环形缓冲区事实上就是一个字节数组。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70 3][]

c）环形缓冲区首先会定义一个分割线，然后从该分割线出发，一侧写元数据，一侧写真实数据，，待写到80%时，则将缓冲区内容spill（溢出）到溢写文件，同时以20%空白区域的中间位置为分割线，反向写。

d）关于环形缓冲区的几个问题：

*  如果溢写速度慢，反向写速度快，则会覆盖之前的内容吗？答案是不会，如果溢写线程速度慢的话，那么反向写线程会等待。
 *  为什么要环形缓冲区？使用环形缓冲区，便于写入缓冲区和写出缓冲区同时进行。
 *  为什么不等缓冲区满了再spill?会出现阻塞。
 *  数据的分区和排序是在哪完成的？分区是根据元数据meta中的分区号partition来分区的，排序是在spill的时候排序。

e）一个MapTask对应一个环形缓冲区，一个环形缓冲区会生成多个溢写文件，溢写时会进行快速排序，注意：排序时是在内存中对元数据进行排序，这样就不会移动真实数据位置，溢写时则根据排好序的元数据去寻找指定位置的真实数据。

f）在溢写时，可以进行combiner预聚合，但有前提条件。

g）多个溢写文件再进行归并排序，合并成一个溢写文件，多个分区都在这一个溢写文件中，分区内有序，分区间无序。

h）ReduceTask从多个MapTask中拉取对应分区的文件，再归并排序，有序的好处是能快速地让具有相同key的kv对进入reduce方法。

i）事实上，数据量小的情况下都是MapTask全部执行完毕之后再开启ReduceTask，但是当数据量大时就会等待过久，可以设置等一部分MapTask执行完之后就开启ReduceTask从而提前聚合一部分数据。

j）Shuffle中的缓冲区大小会影响到MapReduce 程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快。缓冲区的大小可以通过参数调整，参数：mapreduce.task.io.sort.mb 默认100M。

## 4.3、shuffle(洗牌)机制 ##

Map方法之后，Reduce方法之前的数据处理过程称之为Shuffle。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70 4][]

shuffle过程做的事：分区、排序、合并、压缩。

## 4.4、MapTask工作机制 ##

MapTask分为5个阶段：

1.  Read 阶段
2.  Map 阶段
3.  Collect 收集阶段：在用户编写map()函数中，当数据处理完成后，一般会调用OutputCollector.collect()输出结果。在该函数内部，它会将生成的key/value分区（调用Partitioner），并写入一个环形内存缓冲区中。
4.  Spill 阶段：即溢写，当环形缓冲区满后，MapReduce 会将数据写到本地磁盘上，生成一个临时文件。需要注意的是，将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作 。
5.  Merge阶段 ：当所有数据处理完成后MapTask对所有临时文件进行一次合并，以确保最终只会生成一个数据文件。在进行文件合并过程中， MapTask以分区为单位进行合并。让每个MapTask最终只生成一个数据文件，可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。

溢写阶段详情：

1.  利用快速排序算法对缓存区内的数据进行排序，排序方式是，先按照分区编号Partition进行排序，然后按照key进行排序。这样经过排序后，数据以分区为单位聚集在一起，且同一分区内所有数据按照key有序。
2.  按照分区编号由小到大依次将每个分区中的数据写入临时文件中 。如果 用户设置了Combiner，则写入文件之前，对每个分区中的数据进行一次聚集操作。
3.  将分区数据的元信息写到内存索引数据结构SpillRecord中 ，其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB，则将内存索引写到文件output/spillN.out.index中 。

## 4.5、ReduceTask工作机制 ##

ReduceTask一共3个阶段：

1.  Copy 阶段：ReduceTask从各个MapTask上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。
2.  Sort 阶段：为了将key 相同的数据聚在一起，且由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。
3.  Reduce 阶段

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70]: /images/20221020/376c9b8450044c34a99c37596f44c15f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70 1]: /images/20221020/cdd928f23a7a4dbd963b98c89737e53b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70 2]: /images/20221020/ee93ae69f1d84702b7639b3f5d9fc028.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70 3]: /images/20221020/1f44922b8aa44dec965a8c6dbdf072f0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70 4]: /images/20221020/2cd99a8c202a48e286a085fa0cade019.png