Hadoop入门系列(10)-MapReduce工作流程详解

逃离我推掉我的手 2023-10-10 12:15 6阅读 0赞

## 简化流程 ##

![在这里插入图片描述][20210105121808233.png]

1.  input: 读取输入文件
2.  map： 文件切片，并切片数量执行MapTask任务
3.  shuffle：分区、排序, 并将任务结果写入分区中
4.  reduce：从分区中读取数据，每个分区分配一个ReduceTask
5.  output：数据输出到文件系统

## MapTask工作机制 ##

### 并行度与决定机制 ###

1.  一个job的map阶段并行度由客户端在提交job时决定
2.  每一个split切片分配一个mapTask
3.  默认 切片大小=blocksize
4.  切片时不考虑数据集整体,而是逐个对每一个文件单独切片

### 工作流程 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0_size_16_color_FFFFFF_t_70]

1.  Read(读取)：使用InputFormat读取数据，默认使用TextInputFormat,从输入切片数据中解析为K/V键值对
2.  Map：将解析出的K/V交给用户编写map()函数处理，并产生一系列新的K/V
3.  Collect（收集）：在用户编写map()函数中，当数据处理完成后，一般会调用OutputCollector.collect()输出结果。在该函数内部，它会将生成的key/value分区（调用Partitioner），并写入一个环形内存缓冲区中
4.  Spill（溢写）：当环形缓冲区满后，MapReduce会将数据写到本地磁盘上，生成一个临时文件。需要注意的是，将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作。
5.  Combine：当所有数据处理完成后，MapTask对所有临时文件进行一次合并，以确保最终只会生成一个数据文件

## ReduceTask工作机制 ##

### ReduceTask并行度 ###

ReduceTask的并行度同样影响整个Job的执行并发度和执行效率，但与MapTask的并发数由切片数决定不同，默认为每个分区分发一个Task，也可以直接手动设置。

job.setNumReduceTasks(5);

### 工作流程 ###

1.  Copy阶段：ReduceTask从各个MapTask上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。
2.  Merge阶段：在远程拷贝数据的同时，ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。
3.  Sort阶段：按照MapReduce语义，用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。
4.  Reduce阶段：reduce()函数将计算结果写到HDFS上。

## Shuffle洗牌 ##

### 定义 ###

Map方法之后，Reduce方法之前的数据处理过程称之为Shuffle。shuffle是连接Map和Reduce之间的桥梁，Map的输出要用到Reduce中必须经过shuffle这个环节，shuffle的性能高低直接影响了整个程序的性能和吞吐量。因为在分布式情况下，reduce task需要跨节点去拉取其它节点上的map task结果。这一过程将会产生网络资源消耗和内存，磁盘IO的消耗。通常shuffle分为两部分：Map阶段的数据准备和Reduce阶段的数据拷贝处理。一般将在map端的Shuffle称之为Shuffle Write，在Reduce端的Shuffle称之为Shuffle Read。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0_size_16_color_FFFFFF_t_70 1]

### Partition分区 ###

#### 定义 ####

分区是MapReduce框架的Map阶段进行数据处理之后，将数据写出时需要进行的一项操作，分区的数量决定了ReduceTask的数量，也决定了最终的输出文件有多少个。

#### 默认分区 ####

默认分区是根据key的hashCode对ReduceTasks个数取模得到的。用户没法控制哪个key存储到哪个分区。

@Public
    @Stable
    public class HashPartitioner<K2, V2> implements Partitioner<K2, V2> {
        
        public HashPartitioner() {
        
        }
    
        public void configure(JobConf job) {
        
        }
    
        public int getPartition(K2 key, V2 value, int numReduceTasks) {
        
            return (key.hashCode() & 2147483647) % numReduceTasks;
        }
    }

### 自定义分区 ###

1.  编写分区类

public class MyPartitioner  extends Partitioner<Text, BytesWritable> {
        
        @Override
        public int getPartition(Text text, BytesWritable bytesWritable, int i) {
        
            // 分区逻辑
            return 0;
        }
    }

1.  设置job

// 设置分区类
            job.setPartitionerClass(MyPartitioner.class);
            // 设置ReduceTask数量
            job.setNumReduceTasks(3);

**注意事项**：

*  分区号必须从零开始，逐一累加
 *  如果ReduceTask的数量=1，则不管MapTask端输出多少个分区文件，最终结果都交给这一个ReduceTask，最终也就只会产生一个结果文件 part-r-00000
 *  如果1<ReduceTask的数量<getPartition的结果数，则有一部分分区数据无处安放，会Exception
 *  如果ReduceTask的数量> getPartition的结果数，则会多产生几个空的输出文件part-r-000xx

## 排序 ##

### 概念 ###

MapTask和ReduceTask均会对数据按照key进行排序。该操作属于Hadoop的默认行为。任何应用程序中的数据均会被排序，而不管逻辑上是否需要。

默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。

对于MapTask，它会将处理的结果暂时放到环形缓冲区中，当环形缓冲区使用率达到一定阈值后，再对缓冲区中的数据进行一次**快速排序**，并将这些有序数据溢写到磁盘上，而当数据处理完毕后，它会对磁盘上所有文件进行**归并排序**。

对于ReduceTask，它从每个MapTask上远程拷贝相应的数据文件，如果文件大小超过一定阈值，则溢写磁盘上，否则存储在内存中。如果磁盘上文件数目达到一定阈值，则进行一次归并排序以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则进行一次合并后将数据溢写到磁盘上。当所有数据拷贝完毕后，ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序。

### 排序分类 ###

1.  部分排序  
    MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部有序
2.  全排序  
    最终输出结果只有一个文件，且文件内部有序。实现方式是只设置一个ReduceTask。但该方法在处理大型文件时效率极低，因为一台机器处理所有文件，完全丧失了MapReduce所提供的并行架构
3.  辅助排序  
    在Reduce端对key进行分组。应用于：在接收的key为bean对象时，想让一个或几个字段相同（全部字段比较不相同）的key进入到同一个reduce方法时，可以采用分组排序。
4.  二次排序  
    先按照key中的某个属性（值）进行排序，再按照key中的另一个属性（值）进行排序，这个叫二次排序。在自定义排序过程中，如果compareTo中的判断条件为两个即为二次排序。

### 自定义排序 ###

bean对象做为key传输，需要实现WritableComparable接口重写compareTo方法，就可以实现排序。

## Combiner合并 ##

### 定义 ###

每一个map都会产生大量的本地输出，Combiner的作用就是对map输出的结果先做一次合并，以较少的map和reduce节点中的数据传输量。

1.  Combiner是MR程序中Mapper和Reduce之外的一种组件
2.  Combiner组件的父类就是Reducer
3.  Combiner和Reducer之间的区别在于运行的位置
4.  Combiner的意义就是对每一个MapTask的输出进行局部汇总，以减小网络传输量。
5.  Combiner能够应用的前提是不能影响最终的业务逻辑，而且，Combiner的输出kv应该跟Reducer的输入kv类型要对应起来。

### 自定义的Combiner ###

1.  自定义一个Combiner继承Reducer，重写Reduce方法
2.  在Job驱动类中设置Class

job.setCombinerClass(MyCombiner.class);

[20210105121808233.png]: https://img-blog.csdnimg.cn/20210105121808233.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210105155710844.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/2021010516305632.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0,size_16,color_FFFFFF,t_70