Hadoop离线_MapReduce下

1246-谭同学

发表文章数:140

热门标签

,
首页 » 大数据 » 正文

1 MapTask运行机制详解以及MapTask的并行度

1.1 MapTask运行流程

第一步:读取数据组件InputFormat(默认TextInputFormat)会通过getSplits方法,对输入目录中的文件(输入目录也就是TextInputFormat的Path)进行逻辑切片得到splits。

getSplits方法属于FileInputFormat,该方法返回的就是一个文件有多少个切片,一个切片对应一个maptask的任务。
切片大小如何决定的?
如果一个切片256M,那么一个512M的文件会产生两个切片,两个maptask
如果一个切片是128M,那么一个512M的文件产生四个切片,产生四个maptaks

第二步:将输入文件切分为splits之后,由RecordReader对象(默认LineRecordReader)进行读取,以 /n 作为分隔符,读取一行数据,返回 <key1,value1>(Key1表示每行首字符偏移值,value1表示这一行文本内容)
第三步:返回<key1,value1>后,进入用户自己继承的Mapper类中,执行用户重写的map函数。(RecordReader读取一行这里调用一次)
第四步:map逻辑完之后,将map的每条结果通过context.write进行collect数据收集。在collect中,会先对其进行分区处理,默认使用HashPartitioner。(分区数,也就是Partitioner用户可以根据自己的需要自定义,设置完成后需要在main方法中添加job.setNumReduceTask() 设置reduceTask数,并且自定义分区后无法在本地系统运行代码,必须打包上传到HDFS上运行)
第五步:接下来,会将数据写入内存,内存中这片区域叫做环形缓冲区,缓冲区的作用是批量收集map结果,减少磁盘IO的影响。<key,value> 对以及 Partition 的结果都会被写入缓冲区。当然写入之前,key与value值都会被序列化成字节数组。
    环形缓冲区其实是一个数组,数组中存放着key、value的序列化数据和key、value的元数据信息,包括partition、key的起始位置、value的起始位置以及value的长度。
    缓冲区是有大小限制,默认是 100MB。MapTask输出结果过多可能会撑爆内存,所以需要在一定条件下将缓冲区中的数据临时写入磁盘,这种操作被称为Spill,中文可译为溢写(这个溢写是由单独线程来完成,不影响往缓冲区写map结果的线程)。整个缓冲区有个溢写的比例 spill.percent(这个比例默认是0.8),不会阻止map的结果输出。当缓冲区的数据已经达到阈值(buffer size * spill percent = 100MB * 0.8 = 80MB),溢写线程启动,锁定这80MB的内存,执行溢写过程。Map task的输出结果还可以往剩下的20MB内存中写,互不影响。
环形缓冲区的大小,涉及到MapTask的调优过程。如果内存充足,可以将环形缓冲区的大小调大。
第六步:当溢写线程启动后,需要对这80MB空间内的key做排序(Sort)。排序是MapReduce模型默认的行为,这里的排序也是对序列化的字节做的排序。
    如果设置了Combiner,会在排序后进行,Combiner适用于Reduce的输入<key,value>对与输出<key,value>对类型完全一致,且不影响最终结果的场景,比如累加、最大值等。Combiner会将有相同key的<key,value>对的value加起来,减少溢写到磁盘的数据量。
第七步:合并溢写文件。每次溢写会在磁盘上生成一个临时文件(写之前判断是否有combiner),如果map的输出结果很大,就会有多次这样的溢写发生,磁盘上相应的也就会有多个临时文件存在。当整个数据处理结束之后开始对磁盘中的临时文件进行merge合并,因为最终的文件只有一个,写入磁盘,并且为这个文件提供了一个索引文件,以记录每个reduce对应数据的偏移量。

1.2 MapTask的并行度

MapTask的并行度,就是指代有多少个MapTask

2 ReduceTask运行机制以及ReduceTask的并行度

2.1 ReduceTask运行流程

  1. Copy阶段。Reduce进程启动一些数据copy线程(Fetcher),通过HTTP方式请求maptask获取属于自己的文件。
  2. Merge阶段。和Map端的Merge相同,只是存放的数值是不同Map端Copy过来的。

Copy过来的数据会先放入内存缓冲区中(这里的缓冲区大小要比map端的更为灵活)。merge有三种形式:内存到内存;内存到磁盘;磁盘到磁盘(默认情况下第一种形式不启用)。当内存中的数据量到达一定阈值,就启动内存到磁盘的merge。与map 端类似,这也是溢写的过程(这个过程中如果你设置有Combiner,也是会启用的),然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行,直到没有map端的数据时才结束,然后启动第三种磁盘到磁盘的merge方式生成最终的文件。

  1. 合并排序。把分散的数据合并成一个大的数据后,还会再对合并后的数据排序。
  2. 对排序后的键值对调用reduce方法。key相等的键值<key,value>对调用一次reduce方法,每次调用会产生零个或者多个键值对,最后把这些输出的键值对写入到HDFS文件中

2.2 ReduceTask运行流程图

Hadoop离线_MapReduce下

3 MapReduce的shuffle过程

概述:
shuffle就是指map阶段处理的数据如何传递给reduce阶段(就是分区、排序、规约和分组的操作),是MapReduce框架中最关键的一个流程。
步骤:

  1. Collect阶段。将MapTask的结果输出到默认大小为100M的环形缓冲区,保存的是 <key,value>,Partition分区信息等。
  2. Spill阶段。当内存中的数据量达到一定的阈值的时候,就会将数据写入本地磁盘,在将数据写入磁盘之前需要对数据进行一次排序的操作,如果配置了combiner,还会将有相同分区号和key的数据进行排序。
  3. Merge阶段。把所有溢出的临时文件进行一次合并操作,以确保一个MapTask最终只产生一个中间数据文件。
  4. Copy阶段。ReduceTask启动Fetcher线程到已经完成MapTask的节点上复制一份属于自己的数据,这些数据默认会保存在内存的缓冲区中,当内存的缓冲区达到一定的阀值的时候,就会将数据写到磁盘之上
  5. Merge阶段。在ReduceTask远程复制数据的同时,会在后台开启两个线程对内存到本地的数据文件进行合并操作。
  6. Sort阶段。在对数据进行合并的同时,会进行排序操作,由于MapTask阶段已经对数据进行了局部的排序,ReduceTask只需保证Copy的数据的最终整体有效性即可

4 shuffle阶段数据的压缩机制

概述:
从map阶段输出的数据,都要通过网络拷贝,发送到reduce阶段,这一过程中,涉及到大量的网络IO,如果数据能够进行压缩,那么数据的发送量就会少得多。
最好用的Hadoop压缩算法:
Snappy

在代码中设置压缩:

//设置map阶段的压缩
Configuration configuration = new Configuration();
configuration.set("mapreduce.map.output.compress","true");
configuration.set("mapreduce.map.output.compress.codec","org.apache.hadoop.io.compress.SnappyCodec");

//设置reduce阶段的压缩
configuration.set("mapreduce.output.fileoutputformat.compress","true");
configuration.set("mapreduce.output.fileoutputformat.compress.type","RECORD");
configuration.set("mapreduce.output.fileoutputformat.compress.codec","org.apache.hadoop.io.compress.SnappyCodec");

配置全局的MapReduce压缩:

cd /export/servers/hadoop-2.6.0-cdh5.14.0/etc/hadoop

vim mapred-site.xml

<!--map端输出数据进行压缩-->
<property>
	<name>mapreduce.map.output.compress</name>
	<value>true</value>
</property>
<property>
	<name>mapreduce.map.output.compress.codec</name>
	<value>org.apache.hadoop.io.compress.SnappyCodec</value>
</property>

<!--reduce端输出数据进行压缩-->
<property>
	<name>mapreduce.output.fileoutputformat.compress</name>
	<value>true</value>
</property>
<property>
	<name>mapreduce.output.fileoutputformat.compress.type</name>
	<value>RECORD</value>
</property>
<property>
	<name>mapreduce.output.fileoutputformat.compress.codec</name>
	<value>org.apache.hadoop.io.compress.SnappyCodec</value> 
</property>

未经允许不得转载:作者:1246-谭同学, 转载或复制请以 超链接形式 并注明出处 拜师资源博客
原文地址:《Hadoop离线_MapReduce下》 发布于2020-11-18

分享到:
赞(0) 打赏

评论 抢沙发

评论前必须登录!

  注册



长按图片转发给朋友

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏

Vieu3.3主题
专业打造轻量级个人企业风格博客主题!专注于前端开发,全站响应式布局自适应模板。

登录

忘记密码 ?

您也可以使用第三方帐号快捷登录

Q Q 登 录
微 博 登 录