0


hadoop之MapReduce框架原理

bbf48a5f21c2b1999ba1bd00b528c9e6.jpeg

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)

1FileInputFormat****实现类

思考:在运行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完结!!!!)


本文转载自: https://blog.csdn.net/m0_61469860/article/details/129675247
版权归原作者 小唐同学(๑>؂<๑) 所有, 如有侵权,请联系我们删除。

“hadoop之MapReduce框架原理”的评论:

还没有评论