Hadoop 之 Shuffle and Sort
2016-05-31 17:20
441 查看
Mapreduce 确保每个reducer的input都是按照key 排序的。系统将 map ouputs 变成 reduce inputs输入的过程被称为 shuffle。shuffle是Mapreduce的 心脏,是奇迹发生的地方。
![](http://img.blog.csdn.net/20160530215450619)
每个map任务都有一个环形的内存缓冲区用于存储任务输出。缓冲区默认大小为100M(可以通过mapreduce.task.io.sort.mb 设置)。一旦缓冲的内容达到阈值(mapre
duce.map.sort.spill.percent,默认未0.80),会开启一个后台线程,将内容 spill(溢出)到磁盘。map的输出继续写到缓冲区,但如果在此期间缓冲区被填满,map会被阻塞直到写磁盘过程完成。溢出写过程按轮询方式将缓冲区内容写到 mapreduce.cluster.local.dir 属性指定的目录中。
在写入磁盘前,线程首先根据最终要输出的reducer将数据分区。在每个分区中,后台线程会在内存中按key 执行排序,如果有一个combiner,它就在排序后的输出上运行。运行combine函数会使map输出更加紧凑,减少写到磁盘的数据和传递给reducer的数据。
每次 buffer达到阈值后,都会新建一个spill文件。因此在map写完其最后一个输出记录后,会有几个溢出文件。在task完成之前,spil文件被合并成一个已分区且已排序的输出文件。apreduce.task.io.sort.factor属性控制着一次最多能合并多少流,默认值为0.
如果至少存在3个spill文件(mapreduce.map.combine.minspills 可以配置),这output写入到磁盘之前会再次运行combiner。反复调用combiner不会对最终结果产生影响。如果只有一个或者两个spill 文件,那么就不值得再对map输出调用combiner,所以不会再运行combiner。
将map输出结果压缩后写到磁盘上是个比较好的注意,这样写入磁盘会更快,节约磁盘空间,并且减少传给reducer的数据量。在默认情况下,输出是不压缩的,需要将mapreduce.map.output.compress设置未true。使用的压缩库由mapreduce.map.output.compress.codec属性指定。
reducer 通过HTTP方式得到输出文件的分区,文件分区的工作线程的数量是由属性mapreduce.shuffle.max.threads控制的。此设置是针对每一个NodeManager,不是针对每个map任务。默认情况下这个值是0到处理器数量的2倍。
reduce如何知道要从哪台机器取得map输出的呢?
map任务成功完成后,它会通过 heartbeat机制通知Application Master。因此,对于一个作业,Application Master知道 map output和hosts 之间的映射。reducer会有一个线程轮询master获取map输出的位置,直至获取所有输出的位置。
由于第一个reducer可能失败,所以master并不会在第一reducer检索到map outputs时就从磁盘上删除它们。在job完成后,等待Application Master告知其可以删除时,才得以删除。
如果map任务的输出足够小,会被拷贝到reducer的JVM内存中(mapreduce.reduce.shuffle.input.buffer.percent 属性可以设置其大小,指定用于此用途的堆空间的百分比),否则,将会被复制到磁盘。一旦达到缓冲区大小的阈值时(mapreduce.reduce.shuffle.merge.percent )或者达到map输出的阈值(mapreduce.reduce.merge.inmem.threshold),则合并后spill到磁盘。如果指定combiner,则在合并期间运行它,降低写入硬盘的数据量。
注:为了合并,压缩的文件在内存中都会被解压缩,然后合并。
复制完所有的map输出后,reduce任务进入排序阶段(准确的说是合并阶段,排序是在map端发生的),这个阶段维持map输出的顺序,合并 map outputs。举个例子:
如果有50个map outputs,merge factor是10(mapreduce.task.io.sort.factor 可以设置大小),合并将进行5次,每10个文件合并成一个文件,最后有5个中间文件。
最后将这个5个文件合并未一个文件是调用的reduce函数。
![](http://img.blog.csdn.net/20160531162132869)
合并的文件数实际上并不会像上面例子中所展示的那样。比如上图展示了40个文件合并的次数。
mapred.child.java.opts 可以设置map和reduce任务JVM的大小。任务节点上的内存应该尽量大。
在map端,应该尽量避免多次spill 数据到磁盘来获取性能的提升,一次是最佳的情况。如果可以估算map输出的大小,可以合理的设置mapreduce.task.io.sort.* 属性尽可能减少spill的次数。Mapreduce 计数器计算在作业运行整个阶段中spill到磁盘的record数量,这有利于性能调优。
在reduce端,中间数据全部存储在内存中时,就能获取最佳性能。默认情况下,是不可能发生的。如果reduce函数需要的内存不大,可以把mapreduce.reduce.merge.inmem.threshold设置未0,mapreduce.reduce.input.buffer.percent设置未1.0 就可以提升性能。
Hadoop默认使用的缓冲区大小为 4KB,可以通过属性io.file.buffer.size设置。
Map side
当 map函数开始产生输出时,并不简单的将它写到磁盘。它利用buffer的方式写到内存,并除以效率的考虑,进行预排序。每个map任务都有一个环形的内存缓冲区用于存储任务输出。缓冲区默认大小为100M(可以通过mapreduce.task.io.sort.mb 设置)。一旦缓冲的内容达到阈值(mapre
duce.map.sort.spill.percent,默认未0.80),会开启一个后台线程,将内容 spill(溢出)到磁盘。map的输出继续写到缓冲区,但如果在此期间缓冲区被填满,map会被阻塞直到写磁盘过程完成。溢出写过程按轮询方式将缓冲区内容写到 mapreduce.cluster.local.dir 属性指定的目录中。
在写入磁盘前,线程首先根据最终要输出的reducer将数据分区。在每个分区中,后台线程会在内存中按key 执行排序,如果有一个combiner,它就在排序后的输出上运行。运行combine函数会使map输出更加紧凑,减少写到磁盘的数据和传递给reducer的数据。
每次 buffer达到阈值后,都会新建一个spill文件。因此在map写完其最后一个输出记录后,会有几个溢出文件。在task完成之前,spil文件被合并成一个已分区且已排序的输出文件。apreduce.task.io.sort.factor属性控制着一次最多能合并多少流,默认值为0.
如果至少存在3个spill文件(mapreduce.map.combine.minspills 可以配置),这output写入到磁盘之前会再次运行combiner。反复调用combiner不会对最终结果产生影响。如果只有一个或者两个spill 文件,那么就不值得再对map输出调用combiner,所以不会再运行combiner。
将map输出结果压缩后写到磁盘上是个比较好的注意,这样写入磁盘会更快,节约磁盘空间,并且减少传给reducer的数据量。在默认情况下,输出是不压缩的,需要将mapreduce.map.output.compress设置未true。使用的压缩库由mapreduce.map.output.compress.codec属性指定。
reducer 通过HTTP方式得到输出文件的分区,文件分区的工作线程的数量是由属性mapreduce.shuffle.max.threads控制的。此设置是针对每一个NodeManager,不是针对每个map任务。默认情况下这个值是0到处理器数量的2倍。
reduce side
reduce任务需要集群上若干个map任务的输出作为其特殊的分区文件。每个map任务的完成时间可能不同,因此只要有一个任务完成,reduce任务旧开始复制其输出。reduce任务有少量的copier线程,因此你呢狗狗并行的取得map的输出。默认为5个线程,并且可以通过mapreduce.reduce.shuffle.parallelcopies 属性设置。reduce如何知道要从哪台机器取得map输出的呢?
map任务成功完成后,它会通过 heartbeat机制通知Application Master。因此,对于一个作业,Application Master知道 map output和hosts 之间的映射。reducer会有一个线程轮询master获取map输出的位置,直至获取所有输出的位置。
由于第一个reducer可能失败,所以master并不会在第一reducer检索到map outputs时就从磁盘上删除它们。在job完成后,等待Application Master告知其可以删除时,才得以删除。
如果map任务的输出足够小,会被拷贝到reducer的JVM内存中(mapreduce.reduce.shuffle.input.buffer.percent 属性可以设置其大小,指定用于此用途的堆空间的百分比),否则,将会被复制到磁盘。一旦达到缓冲区大小的阈值时(mapreduce.reduce.shuffle.merge.percent )或者达到map输出的阈值(mapreduce.reduce.merge.inmem.threshold),则合并后spill到磁盘。如果指定combiner,则在合并期间运行它,降低写入硬盘的数据量。
注:为了合并,压缩的文件在内存中都会被解压缩,然后合并。
复制完所有的map输出后,reduce任务进入排序阶段(准确的说是合并阶段,排序是在map端发生的),这个阶段维持map输出的顺序,合并 map outputs。举个例子:
如果有50个map outputs,merge factor是10(mapreduce.task.io.sort.factor 可以设置大小),合并将进行5次,每10个文件合并成一个文件,最后有5个中间文件。
最后将这个5个文件合并未一个文件是调用的reduce函数。
合并的文件数实际上并不会像上面例子中所展示的那样。比如上图展示了40个文件合并的次数。
配置调优
总的原则是给shuffle过程尽量多的提供内存空间。然而,有一个权衡,需要确保map和reduce能得到足够的内存去执行任务。这个也是编写map函数和reduce函数时尽量少用内存的原因,或者不应该大量或者无限制的使用内存。mapred.child.java.opts 可以设置map和reduce任务JVM的大小。任务节点上的内存应该尽量大。
在map端,应该尽量避免多次spill 数据到磁盘来获取性能的提升,一次是最佳的情况。如果可以估算map输出的大小,可以合理的设置mapreduce.task.io.sort.* 属性尽可能减少spill的次数。Mapreduce 计数器计算在作业运行整个阶段中spill到磁盘的record数量,这有利于性能调优。
在reduce端,中间数据全部存储在内存中时,就能获取最佳性能。默认情况下,是不可能发生的。如果reduce函数需要的内存不大,可以把mapreduce.reduce.merge.inmem.threshold设置未0,mapreduce.reduce.input.buffer.percent设置未1.0 就可以提升性能。
Hadoop默认使用的缓冲区大小为 4KB,可以通过属性io.file.buffer.size设置。
相关文章推荐
- adb 启动指定的activity
- linux --> Autoconf和Automake使用
- Error:Apostrophe not preceded by \ (
- Bash shell command 速查手册
- Linux下yum的基本使用
- linux select()详解( 二)-- UDP最简实例
- log4j使用总结
- NSBlockOperation - 执行串行任务
- opencv (opencv3.X) IplImage相互转换mat
- OptionSetHelper
- Linux命令文件分离之SeparatePackageTool.sh
- 架构师必备技能覆盖点—大话
- uva 10180 Rope Crisis in Ropeland!
- Linux进程控制块 task_struct
- linux 系统结构
- linux端口号与PID的互相查询
- OpenCV下实现单窗口显示多幅图像的源码!
- Top 10 tips to prepare your Dynamics AX 2012 Go Live
- 【Linux设备驱动】字符设备驱动
- 服务器监控之zabbix监控Tomcat篇