hadoop之MapReduce框架原理

 

目录

MapReduce框架的简单运行机制：

Mapper阶段： 

InputFormat数据输入：

        切片与MapTask并行度决定机制：

job提交过程源码解析：

切片逻辑：

1）FileInputFormat实现类

进行虚拟存储

（1）虚拟存储过程：

Shuffle阶段：

排序：

 Combiner合并：

 ReduceTask阶段:

Reduce Join：

Map Join：

---

MapReduce框架的简单运行机制：

MapReduce是分为两个阶段的，MapperTask阶段，和ReduceTask阶段。（中间有一个Shuffle阶段）

Mapper阶段，可以通过选择什么方式（K,V的选择对应不同的方法）来读取数据，读取后把数据交给Mapper来进行后续的业务逻辑（用户写），让后进入Reduce阶段通过Shuffle来拉取Mapper阶段的数据，让后通过OutputFormat(等方法)来写出（可以是ES,mysql，hbase，文件）

Mapper阶段： 

InputFormat数据输入：

        切片与MapTask并行度决定机制：

MapTask个数，决定了并行度（相当于在生成map集合的过程中有几个人在干活），**（不一定越多越好，当数据量小的时候可能开启的众多MapTask的时间用一个MapTask已经计算完成）

数据块：Block是HDFS物理上把数据分成一块一块。数据块是HDFS存储数据单位。

数据切片：数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。数据切片是MapReduce程序计算输入数据的单位，一个切片会对应启动一个MapTask。

 

job提交过程源码解析：

因为我们找的job提交，所以在job提交函数哪里打个断点，

步入函数后   

ensureState(JobState.DEFINE);  是确保你的状态是正确的（状态不对或者running 都会抛异常）

setUseNewAPI();       处理Hadoop不同版本之间的API兼容

connect();          连接，（客户端需要与集群或者本机连接）

checkSpecs(job); 校验 校验输出路径是否已经创建，是否有参return submitter.submitJobInternal(Job.this, cluster);   核心代码    步入的时候需要点两下，

第一个步入是步入的参数Job  第二个才步入此方法 

这个方法是提交job（在集群模式下，提交的job包含（通过客户端方式把jar包提交给集群），在本地不需要提交jar包，jar在本地是存在的）

还会进行切片，生成切片信息（几个切片就有几个MapTask）

还会 生成xml文件

综上  job提交会交三样东西（jar,xml文件，切片信息---》集群模式下）

最后会删除所有的信息文件

切片逻辑：

**（切片是每一个文件单独切片）

在本地是32m一块，前边说过，默认一块对应一个切片，但是有前提条件，再你减去32m的时候，余下最后一块如果大于1.1倍就重新分配切片，但如果小于1.1，则不能更新分片

例子1：

已有一个32.1m的数据   物理分块是（32m+0.1m）切片分布是（1个切片，因为32.1/32=1.003125<1.1   所以使用一个切片）

例子2：

已有一个100m的数据

100-32-32=36>32(36/32=1.125>1.1   所以最后36m需要分配两个切片)

**块的大小没办法改变，但是可以调切片大小（maxSize让切片调小）（minSize让切片调大）

切片总结：

（开一个MapTask  默认是占1g内存+1个cpu）

  

1）FileInputFormat实现类

思考：在运行MapReduce程序时，输入的文件格式包括：基于行的日志文件、二进制格式文件、数据库表等。那么，针对不同的数据类型，MapReduce是如何读取这些数据的呢？

FileInputFormat常见的接口实现类包括：TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定义InputFormat等。(应用场景的不同选择不同的接口实现类)

TextInputFormat是默认的FileInputFormat实现类。按行读取每条记录。键是存储该行在整个文件中的起始字节偏移量， LongWritable类型。值是这行的内容，不包括任何行终止符（换行符和回车符），Text类型。

CombineTextInputFormat用于小文件过多的场景，它可以将多个小文件从逻辑上规划到一个切片中，这样，多个小文件就可以交给一个MapTask处理。

进行虚拟存储

（1）虚拟存储过程：

将输入目录下所有文件大小，依次和设置的setMaxInputSplitSize（切片大小）值比较，如果不大于设置的最大值，逻辑上划分一个块。如果输入文件大于设置的最大值且大于两倍，那么以最大值切割一块；当剩余数据大小超过设置的最大值且不大于最大值2倍，此时将文件均分成2个虚拟存储块（防止出现太小切片）。

 测试：

再不使用CombineTextInputFormat情况下（默认TextInputFormat）  

 可以看到切片为4

添加代码，设置实现类为CombineTextInputFormat     和   设置虚拟存储切片大小

// 如果不设置InputFormat，它默认用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);//虚拟存储切片最大值设置4m
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);

  

 可以看到，现在是3个切片

我们可以通过改变虚拟切片大小来改变调用的切片的数量

综上：影响切片的数量的因素为：（1）数据量的大小（2）切片的大小（一般会自动调整）（3）文件格式（有些文件是不可切片的）

影响切片大小的因素：   HDFS中块的大小（通过调maxsize,minsize与块的大小进行比较来判断）

Shuffle阶段：

shuffle阶段是一个从mapper阶段出来的后的阶段，会写入（k,v）一个环形缓冲区（缓冲区分为两半，一半存储索引，一半存储数据，默认100m,到达80%后会反向逆写（减少时间消耗，提高效率，逆写是因为不需要等待全部溢写后在进行写入操作）逆写入文件前会进行分区（分区的个数与reduceTask的个数有关）排序（对key进行排序，但是存储位置并不发生改变，只改变索引的位置，改变存储位置消耗资源较大））写入文件后会进行归并排序（在有序的情况下，归并是最高效的））

排序：

排序可以自定义排序，举例全排序：

自定义了一个Bean类，bean对象做为key传输，需要实现WritableComparable接口重写compareTo方法，就可以实现排序。

 Combiner合并：

并不满足所有生产环境下，只有在不影响最终业务逻辑下才可以实现（求和就可以，算平均值就不可以） 

combiner与reducetask区别如下：

 ReduceTask阶段:

（1）Copy阶段：ReduceTask从各个MapTask上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。

（2）Sort阶段：在远程拷贝数据的同时，ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。按照MapReduce语义，用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。

（3）Reduce阶段：reduce()函数将计算结果写到HDFS上。

ReduceTask的个数可以手动进行设置，设置几就会产生几个文件（分区同上）

Reduce Join：

简述流程：

（1）自定义bean对象（序列化反序列化函数---implements Writable）

（2）写mapper类     先重写setup方法（因为本案例需要两个文件，初始化（读多个文 希望先获取到文件名称（多文件） 一个文件一个切片   setup方法是一个优化手段 获取文件名称）

（3）写reduce类（业务逻辑）   先创建一个集合（类型为bean类型）和bean对象用于存储

用for循环遍历value（key是一样的  一样的key才会进入同一个reduce方法）

获取文件名判断写出不同的业务逻辑

"order"表：

先创建一个bean对象，用于存储数据，用于后续写入集合

用到方法   BeanUtils.copyProperties(tmpOrderBean,value);  获取原数据

让后加入上述创建的集合 orderBeans.add(tmpOrderBean);

“pd”表：

BeanUtils.copyProperties(pdBean,value);直接获取原数据

存储结束，结合阶段：

使用增强for

orderbean.setPname(pdBean.getPname());

使用set函数直接设置集合中的pname

让后写入

context.write(orderbean,NullWritable.get());
业务结束

Reduce Join的缺点：这种方式中，合并的操作是在Reduce阶段完成，Reduce端的处理压力太大，Map节点的运算负载则很低，资源利用率不高，且在Reduce阶段极易产生数据倾斜。

Map Join：

使用场景

Map Join适用于一张表十分小、一张表很大的场景。

Map端实现数据合并就解决了Reduce Join的缺点（数据倾斜）

简述流程：

在map类中

setup方法：将较小文件读入缓存，将数据存储到全局的map集合中，将缓存中的数据全部写入

重写的map方法中：

转换成字符串在切割，通过切割后的数组获取map集合中的pname

让后重新设置输出文件的格式进行写出

（至此mapreduce完结！！！！）