IT数码 购物 网址 头条 软件 日历 阅读 图书馆
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓
图片批量下载器
↓批量下载图片,美女图库↓
图片自动播放器
↓图片自动播放器↓
一键清除垃圾
↓轻轻一点,清除系统垃圾↓
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁
 
   -> 大数据 -> Mapreduce的工作流程以及Shuffle机制 -> 正文阅读

[大数据]Mapreduce的工作流程以及Shuffle机制

仅作复习时使用。
MapReduce工作流程
在这里插入图片描述


第一步,准备好文件;

第二步,切片分析;

第三步,客户端会提交3个信息:Job的切片、jar包(集群模式才有)、Job运行相
关的参数信息;

第四步,Yarn会开启一个Mr appmaster(整个任务的老大),Mr appmaster会读
取客户端提交的信息,根据切片信息开启对应个数的MapTask;

后续讲解一个MapTask的工作内容:

第五步,MapTask开启InputFormat(默认TestInputFormat)来按行读取对应切
片内容,生成对应的kv(K是偏移量,V是每一行的内容),然后返回给Mapper;

第六步,读取数据后,调用jar中编写的mapper运行map函数,调用编写的运算逻
辑;

第七步,数据被写入到环形缓冲区中;

(缓冲区中数据对<k,v>从右半边开始写入,索引从左半边开始写入;存储规则按照
keystart---valstart---keystart---valstart...的形式;(keystart和
valuestart之间指向的内容就是key数据,valuestart和keystart之间指向的内
容就是value数据))

环形缓冲区工作详解:
这里之间借鉴了一篇文章,但是链接找不到了,有小伙伴看到了可以提醒我一下:
在这里插入图片描述
环形缓冲区分为三块,空闲区、数据区、索引区。初始位置取名叫做“赤道”,就是圆环上的白线那个位置。初始状态的时候,数据和索引都为0,所有空间都是空闲状态。
tips:这里有一个调优参数,可以设置环形缓冲区的大小:
mapreduce.task.io.sort.mb,默认100M,可以稍微设置大一些,但不要太大,因为每个spilt就128M。
在这里插入图片描述
环形缓冲区写入的时候,有个细节:数据是从赤道的右边开始写入,索引(每次申请4kb)是从赤道是左边开始写。这个设计很有意思,这样两个文件各是各的,互不干涉。
在这里插入图片描述
在数据和索引的大小到了mapreduce.map.sort.spill.percent参数设置的比例时(默认80%,这个是调优的参数),会有两个动作:

1、对写入的数据进行原地排序,并把排序好的数据和索引spill到磁盘上去;
2、在空闲的20%区域中,重新算一个新的赤道,然后在新赤道的右边写入数据,左边写入索引;
3、当20%写满了,但是上一次80%的数据还没写到磁盘的时候,程序就会panding一下,等80%空间腾出来之后再继续写。(也可以理解成等待上一次写入的80%数据全部刷写完成后)
如此循环往复,永不停歇,直到所有任务全部结束。整个操作都在内存,形状像一个环,所以才叫环形缓冲区。

环形缓冲区的设计目的:

环形缓冲区不需要重新申请新的内存,始终用的都是这个内存空间。大家知道MR是用
java写的,而Java有一个最讨厌的机制就是Full GC。Full GC总是会出来捣乱,这个bug也
非常隐蔽,发现了也不好处理。环形缓冲区从头到尾都在用那一个内存,不断重复利
用,因此完美的规避了Full GC导致的各种问题,同时也规避了频繁申请内存引发的其他问题。   

第八步中,排序方法是快排,排序指的是在溢写前排序,而不是写进去的时候就排序;排序针对的对象不是数据本身,而是数据对应的索引,而后按照字典的顺序排;(不同分区中的数据会被分配到不同的reduce任务中);分区数决定ReduceTask;
第九步,把排好序的内容一些到磁盘文件中;
第十步,对溢写文件做归并排序(即图中左半部分和右半部分文件分别做归并排序);
第十一步,这一步不是一定发生的,这是一个预聚合的过程,即提前进行聚合然后再归并排序,相当于是特殊预聚合情况下的第十步;
在这里插入图片描述
第十二步,所有MapTask任务完成后,启动相应数量的ReduceTask(数目等于设定的分区数)(不是绝对的,有可能先有一部分MapTask先进行了Reduce过程,可进行配置)(这一步可以不说明,直接说明Reducer任务会主动从Mapper任务复制其输出的键值对即可)
第十三步,ReduceTask拿到数据后进行合并,然后再进行一次归并排序(不同MapTask的同一分区数据放到一起做归并排序),这样后续就能很容易把相同key的内容发送到1个Reduce方法中;
第十四步,一次读取一组内容到Reduce方法中(也即key相同的一组数据);
第十五步,这里不做讲解;
第十六步,由OutputFormat向外写出数据;

Shuffle机制
Map方法之后,Reduce方法之前的数据处理过程称之为shuffle;
在这里插入图片描述
Map端的shuffle
一、Map方法出来之后,首先进入getPartition方法,标记数据的分区;
二、进入环形缓冲区,存放数据和索引,然后溢写到磁盘,再溢写之前有一个快排过程;其中排序方法是快排,排序对象是索引,排序顺序是按字典顺序排;
三、溢写文件包括spill.index和spill.out,前者为索引文件,后者为实际kv数据;(combiner为可选过程)Reduce下载数据时的依据就是通过这个index文件来判断;
四、对溢写出来的文件进行归并排序,这里也有可能产生combiner过程,也能设置压缩过程,减少IO消耗;
五、压缩后的数据写到磁盘中,等待ReduceTask的拉取;
Reduce端的shuffle
 Reduce端的shuffle主要包括三个阶段,copy、sort(merge)和reduce。
(1)Copy阶段:Reducer通过Http方式得到输出文件的分区。
  (reduce端可能从n个map的结果中获取数据)ReduceTask会把Mapper端的输出数据复制过来,先把同一分区(对应自身分区)数据拉到内存中,如果内存不够则移除到磁盘中。
(2)sort阶段:
  拉取好之后,对内存和磁盘中所有的数据再做合并和归并排序(把不同map阶段输出的同一分区的数据再做归并排序),然后按照key分组,同组数据进入到Reduce方法中;(内存溢出则做合并保存到磁盘,磁盘中的文件过多做归并排序,分组排序不是一定发生的)
(3)Reducer的参数:最后将合并后的结果作为输入传入Reduce任务中。(一般来说是等待Map阶段完成后才会开始reduce方法)
最后就是Reduce过程了,执行Reducer中的计算逻辑,并将结写到HDFS上。

分区规则:最后不同结果在不同文件中的划分依据是:key.hashcode() % 
numReduceTasks的余数,余数相同的会被划分在一起;(所以是按照key的哈希值
分区;若numReduceTasks=1,直接输出在一个文件内;
    
job.setNumReduceTasks(2);//设置ReduceTask数目,也即分区数;    

排序总结:

? 环形缓冲区的排序是在溢写前排序,而不是写进去的时候就排序,排序方法是快排;排序针对的对象不是数据本身,而是数据对应的索引,而后按照字典的顺序排;溢出到文件之后,会对<k,v>进行归并排序(把key相同的放在一起);不同分区中的用数据会被分配到不同的reduce任务中;

? reduce阶段会进行一次合并文件、归并排序(同一分区的排在一起);(还可能有一个分组排序,但现在用得不多)

? 排序是MapReduce框架中最重要的操作之一。MapTask和ReduceTask都会对数据按照key进行排序。该操作属于Hadoop的默认行为。任何应用程序的数据均会被排序,而不管逻辑上是否需要。默认是按照字典顺序排序,且实现该排序的方法是快速排序

为什么要在map阶段排序?

提前排好序,这样可以直接加快效率,否则reduce需要对所有的<k,v>单独进行判断,大大降低了效率;

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

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

补充内容:
小文件处理机制,这里只讲解其中一个,即CombineTextInputFormat;

由于框架默认的TextInputFormat切片机制是对文件进行规划切片的,不管文件多小,都会是一个单独的切片,都会交给一个MapTask,这样如果有大量小文件,就会产生大量的MapTask**,而每一个MapTask都会占用默认1G的内存和一个cpu,这样集群处理效率会极其低下,甚至崩溃;

应用场景:CombineTextInputFormat切片机制用于小文件过多的场景,它将多个小文件从逻辑上规划到一个切片中,这样多个小文件就交给一个MapTask处理;

设置虚拟存储切片最大值:Combine TextInputFormat.setMaxInputSplitSize(Job,4194304);(4M)

切片机制:虚拟存储过程和切片过程两部分;

虚拟存储过程:核心:文件大小大于setMaxInputSplitSize默认值的,进行拆分,拆到刚好每一份都<=setMaxInputSplitSize默认值为止;所有文件拆好后,两两合并,如果此时>=setMaxInputSplitSize默认值则合并为一个切片,如果还小于则继续和下一个存储文件合并,直到大于默认值或是全部合并完为止;

优化:(了解)

1)Map 阶段
(1)增大环形缓冲区大小。由 100m 扩大到 200m (2)增大环形缓冲区溢写的比
例。由 80%扩大到 90%3)减少对溢写文件的 merge 次数。(10 个文件,一次 
20 个 merge) (4)不影响实际业务的前提下,采用 Combiner 提前合并,减少
 I/O。
    
2)Reduce 阶段
(1)合理设置 Map 和 Reduce 数:两个都不能设置太少,也不能设置太多。太
少,会导致 Task 等待,延长处理时间;太多,会导致 Map、Reduce 任务间竞争
资源,造成处理超时等错误。
(2)设置 Map、Reduce 共存:调整 slowstart.completedmaps 参数,使 Map 
运行到一定程度后,Reduce 也开始运行,减少 Reduce 的等待时间。
(3)规避使用 Reduce,因为 Reduce 在用于连接数据集的时候将会产生大量的网络消耗。
(4)增加每个 Reduce 去 Map 中拿数据的并行数
(5)集群性能可以的前提下,增大 Reduce 端存储数据内存的大小。
 
3)IO 传输
采用数据压缩的方式,减少网络 IO 的的时间。安装 Snappy 和 LZOP 压缩编码器。
压缩:
(1)map 输入端主要考虑数据量大小和切片,支持切片的有 Bzip2、LZO。注意:
LZO 要想支持切片必须创建
索引;
(2)map 输出端主要考虑速度,速度快的 snappy、LZO;
(3)reduce 输出端主要看具体需求,例如作为下一个mr输入需要考虑切片,永久
保存考虑压缩率比较大的 gzip。
    
4)整体
(1)NodeManager 默认内存 8G,需要根据服务器实际配置灵活调整,例如 128G 内存,配置为 100G 内存左右,yarn.nodemanager.resource.memory-mb。 
(2)单任务默认内存 8G,需要根据该任务的数据量灵活调整,例如 128m 数据,配置 1G 内存,
yarn.scheduler.maximum-allocation-mb。 
(3)mapreduce.map.memory.mb :控制分配给MapTask 内存上限,如果超过会
kill掉进程(报错:Container is running beyond physical memory 
limits. Current usage:565MB of512MB physical memory used;Killing 
Container)。默认内存大小为 1G,如果数据量是 128m,正常不需要调整内存;
如果数据量大于 128m,可以增加 MapTask 内存,最大可以增加到 4-5g。 
(4)mapreduce.reduce.memory.mb:控制分配给ReduceTask内存上限。默认内
存大小为1G,如果数据量是128m,正常不需要调整内存;如果数据量大于 128m,可
以增加 ReduceTask 内存大小为 4-5g。 
(5)mapreduce.map.java.opts:控制 MapTask 堆内存大小。(如果内存不够,报:java.lang.OutOfMemoryError) 
(6)mapreduce.reduce.java.opts:控制 ReduceTask 堆内存大小。(如果内存不够,报:java.lang.OutOfMemoryError) 
(7)可以增加 MapTask 的 CPU 核数,增加 ReduceTask 的 CPU 核数
(8)增加每个 Container 的 CPU 核数和内存大小
(9)在 hdfs-site.xml 文件中配置多目录(多磁盘)  
(10)NameNode 有一个工作线程池,用来处理不同 DataNode 的并发心跳以及客户端并发的元数据操作。
比如集群规模为 8 台时,此参数设置为 41。参数设置方法略;
  大数据 最新文章
实现Kafka至少消费一次
亚马逊云科技:还在苦于ETL?Zero ETL的时代
初探MapReduce
【SpringBoot框架篇】32.基于注解+redis实现
Elasticsearch:如何减少 Elasticsearch 集
Go redis操作
Redis面试题
专题五 Redis高并发场景
基于GBase8s和Calcite的多数据源查询
Redis——底层数据结构原理
上一篇文章      下一篇文章      查看所有文章
加:2021-07-25 00:00:03  更:2021-07-25 00:00:07 
 
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁

360图书馆 购物 三丰科技 阅读网 日历 万年历 2024年4日历 -2024/4/30 19:28:16-

图片自动播放器
↓图片自动播放器↓
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓
一键清除垃圾
↓轻轻一点,清除系统垃圾↓
图片批量下载器
↓批量下载图片,美女图库↓
  网站联系: qq:121756557 email:121756557@qq.com  IT数码