Shuffle的正常意思是洗牌或弄乱,可能大家更熟悉的是Java API里的Collections.shuffle(List)方法,它会随机地打乱参数list里的元素顺序。如果你不知道MapReduce里Shuffle是什么,那么请看这张图:

这张是官方对Shuffle过程的描述。但我可以肯定的是,单从这张图你基本不可能明白Shuffle的过程,因为它与事实相差挺多,细节也是错乱的。后面我会具体描述Shuffle的事实情况,所以这里你只要清楚Shuffle的大致范围就成-怎样把map task的输出结果有效地传送到reduce端。也可以这样理解, Shuffle描述着数据从map task输出到reduce task输入的这段过程。

在Hadoop这样的集群环境中,大部分map task与reduce task的执行是在不同的节点上。当然很多情况下Reduce执行时需要跨节点去拉取其它节点上的map task结果。如果集群正在运行的job有很多,那么task的正常执行对集群内部的网络资源消耗会很严重。这种网络消耗是正常的,我们不能限制,能做的就是最大化地减少不必要的消耗。还有在节点内,相比于内存,磁盘IO对job完成时间的影响也是可观的。从最基本的要求来说,我们对Shuffle过程的期望可以有:

  • 完整地从map task端拉取数据到reduce 端。
  • 在跨节点拉取数据时,尽可能地减少对带宽的不必要消耗。
  • 减少磁盘IO对task执行的影响。

OK,看到这里时,大家可以先停下来想想,如果是自己来设计这段Shuffle过程,那么你的设计目标是什么。我想能优化的地方主要在于减少拉取数据的量及尽量使用内存而不是磁盘。

我的分析是基于Hadoop0.21.0的源码,如果与你所认识的Shuffle过程有差别,不吝指出。我会以WordCount为例,并假设它有8个map task和3个reduce task。从上图看出,Shuffle过程横跨map与reduce两端,所以下面我也会分两部分来展开。

先看看map端的情况,如下图:

上图可能是某个map task的运行情况。拿它与官方图的左半边比较,会发现很多不一致。官方图没有清楚地说明partition, sort与combiner到底作用在哪个阶段。我画了这张图,希望让大家清晰地了解从map数据输入到map端所有数据准备好的全过程。

整个流程我分了四步。简单些可以这样说,每个map task都有一个内存缓冲区,存储着map的输出结果,当缓冲区快满的时候需要将缓冲区的数据以一个临时文件的方式存放到磁盘,当整个map task结束后再对磁盘中这个map task产生的所有临时文件做合并,生成最终的正式输出文件,然后等待reduce task来拉数据。

当然这里的每一步都可能包含着多个步骤与细节,下面我对细节来一一说明: 
1.        在map task执行时,它的输入数据来源于HDFS的block,当然在MapReduce概念中,map task只读取split。Split与block的对应关系可能是多对一,默认是一对一。在WordCount例子里,假设map的输入数据都是像“aaa”这样的字符串。

2.        在经过mapper的运行后,我们得知mapper的输出是这样一个key/value对: key是“aaa”, value是数值1。因为当前map端只做加1的操作,在reduce task里才去合并结果集。前面我们知道这个job有3个reduce task,到底当前的“aaa”应该交由哪个reduce去做呢,是需要现在决定的。

MapReduce提供Partitioner接口,它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力,如果用户自己对Partitioner有需求,可以订制并设置到job上。

在我们的例子中,“aaa”经过Partitioner后返回0,也就是这对值应当交由第一个reducer来处理。接下来,需要将数据写入内存缓冲区中,缓冲区的作用是批量收集map结果,减少磁盘IO的影响。我们的key/value对以及Partition的结果都会被写入缓冲区。当然写入之前,key与value值都会被序列化成字节数组。

整个内存缓冲区就是一个字节数组,它的字节索引及key/value存储结构我没有研究过。如果有朋友对它有研究,那么请大致描述下它的细节吧。

3.        这个内存缓冲区是有大小限制的,默认是100MB。当map task的输出结果很多时,就可能会撑爆内存,所以需要在一定条件下将缓冲区中的数据临时写入磁盘,然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为Spill,中文可译为溢写,字面意思很直观。这个溢写是由单独线程来完成,不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出,所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8,也就是当缓冲区的数据已经达到阈值(buffer size * spill percent = 100MB * 0.8 = 80MB),溢写线程启动,锁定这80MB的内存,执行溢写过程。Map task的输出结果还可以往剩下的20MB内存中写,互不影响。

当溢写线程启动后,需要对这80MB空间内的key做排序(Sort)。排序是MapReduce模型默认的行为,这里的排序也是对序列化的字节做的排序。

在这里我们可以想想,因为map task的输出是需要发送到不同的reduce端去,而内存缓冲区没有对将发送到相同reduce端的数据做合并,那么这种合并应该是体现是磁盘文件中的。从官方图上也可以看到写到磁盘中的溢写文件是对不同的reduce端的数值做过合并。所以溢写过程一个很重要的细节在于,如果有很多个key/value对需要发送到某个reduce端去,那么需要将这些key/value值拼接到一块,减少与partition相关的索引记录。

在针对每个reduce端而合并数据时,有些数据可能像这样:“aaa”/1, “aaa”/1。对于WordCount例子,就是简单地统计单词出现的次数,如果在同一个map task的结果中有很多个像“aaa”一样出现多次的key,我们就应该把它们的值合并到一块,这个过程叫reduce也叫combine。但MapReduce的术语中,reduce只指reduce端执行从多个map task取数据做计算的过程。除reduce外,非正式地合并数据只能算做combine了。其实大家知道的,MapReduce中将Combiner等同于Reducer。

如果client设置过Combiner,那么现在就是使用Combiner的时候了。将有相同key的key/value对的value加起来,减少溢写到磁盘的数据量。Combiner会优化MapReduce的中间结果,所以它在整个模型中会多次使用。那哪些场景才能使用Combiner呢?从这里分析,Combiner的输出是Reducer的输入,Combiner绝不能改变最终的计算结果。所以从我的想法来看,Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致,且不影响最终结果的场景。比如累加,最大值等。Combiner的使用一定得慎重,如果用好,它对job执行效率有帮助,反之会影响reduce的最终结果。

4.        每次溢写会在磁盘上生成一个溢写文件,如果map的输出结果真的很大,有多次这样的溢写发生,磁盘上相应的就会有多个溢写文件存在。当map task真正完成时,内存缓冲区中的数据也全部溢写到磁盘中形成一个溢写文件。最终磁盘中会至少有一个这样的溢写文件存在(如果map的输出结果很少,当map执行完成时,只会产生一个溢写文件),因为最终的文件只有一个,所以需要将这些溢写文件归并到一起,这个过程就叫做Merge。Merge是怎样的?如前面的例子,“aaa”从某个map task读取过来时值是5,从另外一个map 读取时值是8,因为它们有相同的key,所以得merge成group。什么是group。对于“aaa”就是像这样的:{“aaa”, [5, 8, 2, …]},数组中的值就是从不同溢写文件中读取出来的,然后再把这些值加起来。请注意,因为merge是将多个溢写文件合并到一个文件,所以可能也有相同的key存在,在这个过程中如果client设置过Combiner,也会使用Combiner来合并相同的key。

至此,map端的所有工作都已结束,最终生成的这个文件也存放在TaskTracker够得着的某个本地目录内。每个reduce task不断地通过RPC从JobTracker那里获取map task是否完成的信息,如果reduce task得到通知,获知某台TaskTracker上的map task执行完成,Shuffle的后半段过程开始启动。

简单地说,reduce task在执行之前的工作就是不断地拉取当前job里每个map task的最终结果,然后对从不同地方拉取过来的数据不断地做merge,也最终形成一个文件作为reduce task的输入文件。见下图:

如map 端的细节图,Shuffle在reduce端的过程也能用图上标明的三点来概括。当前reduce copy数据的前提是它要从JobTracker获得有哪些map task已执行结束,这段过程不表,有兴趣的朋友可以关注下。Reducer真正运行之前,所有的时间都是在拉取数据,做merge,且不断重复地在做。如前面的方式一样,下面我也分段地描述reduce 端的Shuffle细节: 
1.        Copy过程,简单地拉取数据。Reduce进程启动一些数据copy线程(Fetcher),通过HTTP方式请求map task所在的TaskTracker获取map task的输出文件。因为map task早已结束,这些文件就归TaskTracker管理在本地磁盘中。

2.        Merge阶段。这里的merge如map端的merge动作,只是数组中存放的是不同map端copy来的数值。Copy过来的数据会先放入内存缓冲区中,这里的缓冲区大小要比map端的更为灵活,它基于JVM的heap size设置,因为Shuffle阶段Reducer不运行,所以应该把绝大部分的内存都给Shuffle用。这里需要强调的是,merge有三种形式:1)内存到内存  2)内存到磁盘  3)磁盘到磁盘。默认情况下第一种形式不启用,让人比较困惑,是吧。当内存中的数据量到达一定阈值,就启动内存到磁盘的merge。与map 端类似,这也是溢写的过程,这个过程中如果你设置有Combiner,也是会启用的,然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行,直到没有map端的数据时才结束,然后启动第三种磁盘到磁盘的merge方式生成最终的那个文件。

3.        Reducer的输入文件。不断地merge后,最后会生成一个“最终文件”。为什么加引号?因为这个文件可能存在于磁盘上,也可能存在于内存中。对我们来说,当然希望它存放于内存中,直接作为Reducer的输入,但默认情况下,这个文件是存放于磁盘中的。至于怎样才能让这个文件出现在内存中,之后的性能优化篇我再说。当Reducer的输入文件已定,整个Shuffle才最终结束。然后就是Reducer执行,把结果放到HDFS上。

----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

我们按照图中的1234步逐步进行说明:
①在map端首先接触的是InputSplit,在InputSplit中含有DataNode中的数据,每一个InputSplit都会分配一个Mapper任务。
②当key/value被写入缓冲区之前,都会被序列化为字节流。mapreduce提供Partitioner接口,它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理(分区)。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力,如果用户自己对Partitioner有需求,可以订制并设置到job上。

注意:虽然Partitioner接口会计算出一个值来决定某个输出会交给哪个reduce去处理,但是在缓冲区中并不会实现物理上的分区,而是将结果加载key-value后面。物理上的分区实在磁盘上进行的。

每个map有一个环形内存缓冲区,用于存储任务的输出。默认大小100MB(io.sort.mb属性)。
③一旦达到阀值80%(io.sort.spil l.percent),一个后台线程就把内容写到(spill:溢写)Linux本地磁盘中的指定目录(mapred.local.dir)下的新建的一个溢出写文件。在这一步会执行两个操作排序和Combiner(前提是设置了Combiner)。

这里大家可能会出现疑问:是将哪部分溢写到磁盘上那?答案是,溢写线程启动时,会锁定这80M的内存,执行溢写过程。而剩余的那20M缓冲区会继续接收map的输出,直到缓冲区写满,Map 才会被阻塞直到spill 完成。spill操作和接收map输出的操作是两个独立的线程,故互不影响。

spill 线程在把缓冲区的数据写到磁盘前,会对它进行一个二次快速排序,首先根据数据所属的partition (分区)排序,然后每个partition 中再按Key 排序。输出包括一个索引文件和数据文件。如果设定了Combiner,将在排序输出的基础上运行。Combiner 就是一个简单Reducer操作,它在执行Map 任务的节点本身运行,先对Map 的输出做一次简单Reduce,使得Map 的输出更紧凑,更少的数据会被写入磁盘和传送到Reducer。spill 文件保存在由mapred.local.dir指定的目录中,map 任务结束后删除。

每次溢写会在磁盘上生成一个溢写文件,如果map的输出结果很大,有多次这样的溢写发生,磁盘上相应的就会有多个溢写文件存在。而如果map的输出很小以至于最终也没有到达阀值,那最后会将其缓冲区的内容写入磁盘。
④因为最终的文件只有一个,所以需要将这些溢写文件归并到一起,
这个过程就叫做Merge。因为merge是将多个溢写文件合并到一个文件,所以可能也有相同的key存在,在这个过程中如果client设置过Combiner,也会使用Combiner来合并相同的key。

从这里我们可以得出,溢写操作是写到了磁盘上,并不一定就是最终的结果,因为最终结果是要只有一个文件,除非其map的输出很小以至于没有没有发生过溢写(也就是说磁盘上只有一个文件)。

到这里,map端的shuffle就全部完成了。

reduce端的shuffle:

map完成后,会通过心跳将信息传给tasktracker,其进而通知jobtracker,reduce task不断地通过RPC从JobTracker那里获取map task是否完成的信息,当得知某个TaskTracker上的map task执行完成,Reduce端的shuffle就开始工作了。

注意:这里是reduce端的shuffle开始工作,而不是reduce操作开始执行,在shuffle阶段reduce不会运行。

同样我们按照图中的标号,分为三个阶段进行讲解。
**①**Copy阶段:reduce端默认有5个数据复制线程从map端复制数据,其通过Http方式得到Map对应分区的输出文件。reduce端并不是等map端执行完后将结果传来,而是直接去map端去Copy输出文件。
**②**Merge阶段:reduce端的shuffle也有一个环形缓冲区,它的大小要比map端的灵活(由JVM的heapsize设置),由Copy阶段获得的数据,会存放的这个缓冲区中,同样,当到达阀值时会发生溢写操作,这个过程中如果设置了Combiner也是会执行的,这个过程会一直执行直到所有的map输出都被复制过来,如果形成了多个磁盘文件还会进行合并,最后一次合并的结果作为reduce的输入而不是写入到磁盘中。
③当Reducer的输入文件确定后,整个Shuffle操作才最终结束。之后就是Reducer的执行了,最后Reducer会把结果存到HDFS上。

转载于:https://www.cnblogs.com/HHR-SUN/p/10478216.html

shuffle的工作原理相关推荐

  1. 基于TensorFolw的人工智能影像诊断平台工作原理解析

    文章对TensorFolw人工智能影像诊断平台的工作原理进行了解析,希望这篇文章能够帮助你更好地理解 Tensorflow. 使用人工智能来辅助病理医生对样本进行诊断,不仅能够大幅度提高医师的诊断效率 ...

  2. Hadoop 4、Hadoop MapReduce的工作原理

    一.MapReduce的概念 MapReduce是hadoop的核心组件之一,hadoop要分布式包括两部分,一是分布式文件系统hdfs,一部是分布式计算框就是mapreduce,两者缺一不可,也就是 ...

  3. MapReduce工作原理图文详解

    前言: 前段时间我们云计算团队一起学习了hadoop相关的知识,大家都积极地做了.学了很多东西,收获颇丰.可是开学后,大家都忙各自的事情,云计算方面的动静都不太大.呵呵~不过最近在胡老大的号召下,我们 ...

  4. 《Hadoop MapReduce性能优化》一1.3 Hadoop MapReduce的工作原理

    本节书摘来异步社区<Hadoop MapReduce性能优化>一书中的第1章,第1.3节,作者: [法]Khaled Tannir 译者: 范欢动 责编: 杨海玲,更多章节内容可以访问云栖 ...

  5. 深入浅出理解 Spark:环境部署与工作原理

    一.Spark 概述 Spark 是 UC Berkeley AMP Lab 开源的通用分布式并行计算框架,目前已成为 Apache 软件基金会的顶级开源项目.Spark 支持多种编程语言,包括 Ja ...

  6. mpls工作原理通俗解释_用这两种方法向最终用户解释NLP模型的工作原理还是不错的...

    点击上方关注,All in AI中国 上周,我看了一个关于"NLP的实践特性工程"的演讲.主要是关于LIME和SHAP在文本分类可解释性方面是如何工作的. 我决定写一篇关于它们的文 ...

  7. Hadoop 之 MapReduce 的工作原理及其倒排索引的建立

    一.Hadoop 简介 下面先从一张图理解MapReduce得整个工作原理 下面对上面出现的一些名词进行介绍 ResourceManager:是YARN资源控制框架的中心模块,负责集群中所有的资源的统 ...

  8. 【hadoop权威指南第四版】第六章MR的工作原理【笔记+代码】

    6.1 运行MR作业 工作原理 四大模块: 客户端,提交MR作业. jobtracker,协调作业的运行.jobtracker 是一个java应用程序,主类是Jobtracker. tasktrack ...

  9. (超详细)MapReduce工作原理及基础编程

    MapReduce工作原理及基础编程(代码见文章后半部分) JunLeon--go big or go home 目录 MapReduce工作原理及基础编程(代码见文章后半部分) 一.MapReduc ...

最新文章

  1. mycat mysql 存储过程_MyCat 学习笔记 第十三篇.数据分片 之 通过HINT执行存储过程...
  2. Windows 动态链接库DLL浅解
  3. # 遍历结构体_C#学习笔记05--枚举/结构体
  4. [hackinglab][CTF][注入关][2020] hackinglab 注入关 writeup
  5. 一个寄存器有几个字节_STM32f103ZET6 学习资料 (连载2 寄存器的操作界限)
  6. vue 父链和子组件索引_解决Vue2.x父组件与子组件之间的双向绑定问题
  7. Python:Numpy库中的invert()函数的用法
  8. ONVIF流媒体播放流程
  9. 启动工程Ehcache报错
  10. 如何使用MASM 5.0汇编语言编译器
  11. 佛系程序员之歌 - 和我一起减减压
  12. 微信公众号 Hash模式授权
  13. ADNI数据集相关概念初步整理
  14. 鸿蒙手机电脑无缝对接,万物互联?华为鸿蒙OS超级终端功能曝光可一键连接附近设备...
  15. 计算机系统组成思维导图
  16. (报错解决)Exception encountered during context initialization
  17. 抖音纸短情长音乐计算机简谱,抖音纸短情长女版谁唱的 纸短情长计算器简谱完整版...
  18. 回忆做嵌入式软件的几年
  19. 亚洲研究院微软笔试题
  20. 2021年新春佳节,《经济学人》是如何报道的?

热门文章

  1. go导入私有仓库中的包配置方法
  2. golang安装:GOPATH环境变量无需配置说明(>1.13)
  3. K8S集群搭建:利用kubeadm构建K8S集群
  4. 【视频】vue指令之@click及其stop修饰符
  5. 简单讲解一下负载均衡、反向代理模式的优点、缺点
  6. MySQL自定义变量的使用
  7. IDEA的查询引用、调用关系图的功能
  8. java遍历本地文件夹_JAVA遍历一个文件夹中的所有文件的小例子
  9. 微信支付分申请接入流程
  10. GitLab修改用户密码