载入中。。。 'S bLog
 
载入中。。。
 
载入中。。。
载入中。。。
载入中。。。
载入中。。。
载入中。。。
 
填写您的邮件地址,订阅我们的精彩内容:


 
hadoop作业map过程调优使用到的参数笔记
[ 2012/3/31 13:24:00 | By: 梦翔儿 ]
 

参数:io.sort.mb(default 100)

当map task开始运算,并产生中间数据时,其产生的中间结果并非直接就简单的写入磁盘。

而是会利用到了内存buffer来进行已经产生的部分结果的缓存,

并在内存buffer中进行一些预排序来优化整个map的性能。

每一个map都会对应存在一个内存buffer,map会将已经产生的部分结果先写入到该buffer中,

这个buffer默认是100MB大小,

但是这个大小是可以根据job提交时的参数设定来调整的,

当map的产生数据非常大时,并且把io.sort.mb调大,

那么map在整个计算过程中spill的次数就势必会降低,

map task对磁盘的操作就会变少,

如果map tasks的瓶颈在磁盘上,这样调整就会大大提高map的计算性能。


参数:o.sort.spill.percent(default 0.80,也就是80%)

map在运行过程中,不停的向该buffer中写入已有的计算结果,

但是该buffer并不一定能将全部的map输出缓存下来,

当map输出超出一定阈值(比如100M),那么map就必须将该buffer中的数据写入到磁盘中去,

这个过程在mapreduce中叫做spill。

map并不是要等到将该buffer全部写满时才进行spill,

因为如果全部写满了再去写spill,势必会造成map的计算部分等待buffer释放空间的情况。

所以,map其实是当buffer被写满到一定程度(比如80%)时,就开始进行spill。

这个阈值也是由一个job的配置参数来控制,

这个参数同样也是影响spill频繁程度,进而影响map task运行周期对磁盘的读写频率的。

但非特殊情况下,通常不需要人为的调整。调整io.sort.mb对用户来说更加方便。


参数:io.sort.factor

当map task的计算部分全部完成后,如果map有输出,就会生成一个或者多个spill文件,这些文件就是map的输出结果。

map在正常退出之前,需要将这些spill合并(merge)成一个,所以map在结束之前还有一个merge的过程。

merge的过程中,有一个参数可以调整这个过程的行为,该参数为:io.sort.factor。

该参数默认为10。它表示当merge spill文件时,最多能有多少并行的stream向merge文件中写入。

比如如果map产生的数据非常的大,产生的spill文件大于10,而io.sort.factor使用的是默认的10,

那么当map计算完成做merge时,就没有办法一次将所有的spill文件merge成一个,而是会分多次,每次最多10个stream。

这也就是说,当map的中间结果非常大,调大io.sort.factor,

有利于减少merge次数,进而减少map对磁盘的读写频率,有可能达到优化作业的目的。


参数:min.num.spill.for.combine(default 3)

当job指定了combiner的时候,我们都知道map介绍后会在map端根据combiner定义的函数将map结果进行合并。

运行combiner函数的时机有可能会是merge完成之前,或者之后,这个时机可以由一个参数控制,

即min.num.spill.for.combine(default 3),当job中设定了combiner,并且spill数最少有3个的时候,

那么combiner函数就会在merge产生结果文件之前运行。

通过这样的方式,就可以在spill非常多需要merge,并且很多数据需要做conbine的时候,

减少写入到磁盘文件的数据数量,同样是为了减少对磁盘的读写频率,有可能达到优化作业的目的。


参数:mapred.compress.map.output(default false)

减少中间结果读写进出磁盘的方法不止这些,还有就是压缩。

也就是说map的中间,无论是spill的时候,还是最后merge产生的结果文件,都是可以压缩的。

压缩的好处在于,通过压缩减少写入读出磁盘的数据量。

对中间结果非常大,磁盘速度成为map执行瓶颈的job,尤其有用。

控制map中间结果是否使用压缩的参数为:mapred.compress.map.output(true/false)。

将这个参数设置为true时,那么map在写中间结果时,就会将数据压缩后再写入磁盘,读结果时也会采用先解压后读取数据。

这样做的后果就是:写入磁盘的中间结果数据量会变少,但是cpu会消耗一些用来压缩和解压。

所以这种方式通常适合job中间结果非常大,瓶颈不在cpu,而是在磁盘的读写的情况。

说的直白一些就是用cpu换IO。

根据观察,通常大部分的作业cpu都不是瓶颈,除非运算逻辑异常复杂。所以对中间结果采用压缩通常来说是有收益的。


参数:mapred.map.output.compression.codec( default org.apache.hadoop.io.compress.DefaultCodec)

当采用map中间结果压缩的情况下,用户还可以选择压缩时采用哪种压缩格式进行压缩,

现在hadoop支持的压缩格式有:GzipCodec,LzoCodec,BZip2Codec,LzmaCodec等压缩格式。

通常来说,想要达到比较平衡的cpu和磁盘压缩比,LzoCodec比较适合。但也要取决于job的具体情况。

用户若想要自行选择中间结果的压缩算法,

可以设置配置参数:mapred.map.output.compression.codec=org.apache.hadoop.io.compress.DefaultCodec或者其他用户自行选择的压缩方式

http://cloud.csdn.net/a/20110121/290650.html

hadoop作业调优参数整理及原理

1 Map side tuning参数

1.1 MapTask运行内部原理

当map task开始运算,并产生中间数据时,其产生的中间结果并非直接就简单的写入磁盘。这中间的过程比较复杂,并且利用到了内存buffer来进行已经产生的部分结果的缓存,并在内存buffer中进行一些预排序来优化整个map的性能。如上图所示,每一个map都会对应存在一个内存buffer(MapOutputBuffer,即上图的buffer in memory),map会将已经产生的部分结果先写入到该buffer中,这个buffer默认是100MB大小,但是这个大小是可以根据job提交时的参数设定来调整的,该参数即为:io.sort.mb。当map的产生数据非常大时,并且把io.sort.mb调大,那么map在整个计算过程中spill的次数就势必会降低,map task对磁盘的操作就会变少,如果map tasks的瓶颈在磁盘上,这样调整就会大大提高map的计算性能。map做sort和spill的内存结构如下如所示:

map在运行过程中,不停的向该buffer中写入已有的计算结果,但是该buffer并不一定能将全部的map输出缓存下来,当map输出超出一定阈值(比如100M),那么map就必须将该buffer中的数据写入到磁盘中去,这个过程在mapreduce中叫做spill。map并不是要等到将该buffer全部写满时才进行spill,因为如果全部写满了再去写spill,势必会造成map的计算部分等待buffer释放空间的情况。所以,map其实是当buffer被写满到一定程度(比如80%)时,就开始进行spill。这个阈值也是由一个job的配置参数来控制,即io.sort.spill.percent,默认为0.80或80%。这个参数同样也是影响spill频繁程度,进而影响map task运行周期对磁盘的读写频率的。但非特殊情况下,通常不需要人为的调整。调整io.sort.mb对用户来说更加方便。

当map task的计算部分全部完成后,如果map有输出,就会生成一个或者多个spill文件,这些文件就是map的输出结果。map在正常退出之前,需要将这些spill合并(merge)成一个,所以map在结束之前还有一个merge的过程。merge的过程中,有一个参数可以调整这个过程的行为,该参数为:io.sort.factor。该参数默认为10。它表示当merge spill文件时,最多能有多少并行的stream向merge文件中写入。比如如果map产生的数据非常的大,产生的spill文件大于10,而io.sort.factor使用的是默认的10,那么当map计算完成做merge时,就没有办法一次将所有的spill文件merge成一个,而是会分多次,每次最多10个stream。这也就是说,当map的中间结果非常大,调大io.sort.factor,有利于减少merge次数,进而减少map对磁盘的读写频率,有可能达到优化作业的目的。

当job指定了combiner的时候,我们都知道map介绍后会在map端根据combiner定义的函数将map结果进行合并。运行combiner函数的时机有可能会是merge完成之前,或者之后,这个时机可以由一个参数控制,即min.num.spill.for.combine(default 3),当job中设定了combiner,并且spill数最少有3个的时候,那么combiner函数就会在merge产生结果文件之前运行。通过这样的方式,就可以在spill非常多需要merge,并且很多数据需要做conbine的时候,减少写入到磁盘文件的数据数量,同样是为了减少对磁盘的读写频率,有可能达到优化作业的目的。

减少中间结果读写进出磁盘的方法不止这些,还有就是压缩。也就是说map的中间,无论是spill的时候,还是最后merge产生的结果文件,都是可以压缩的。压缩的好处在于,通过压缩减少写入读出磁盘的数据量。对中间结果非常大,磁盘速度成为map执行瓶颈的job,尤其有用。控制map中间结果是否使用压缩的参数为:mapred.compress.map.output(true/false)。将这个参数设置为true时,那么map在写中间结果时,就会将数据压缩后再写入磁盘,读结果时也会采用先解压后读取数据。这样做的后果就是:写入磁盘的中间结果数据量会变少,但是cpu会消耗一些用来压缩和解压。所以这种方式通常适合job中间结果非常大,瓶颈不在cpu,而是在磁盘的读写的情况。说的直白一些就是用cpu换IO。根据观察,通常大部分的作业cpu都不是瓶颈,除非运算逻辑异常复杂。所以对中间结果采用压缩通常来说是有收益的。以下是一个wordcount中间结果采用压缩和不采用压缩产生的map中间结果本地磁盘读写的数据量对比:

map中间结果不压缩:

map中间结果压缩:


可以看出,同样的job,同样的数据,在采用压缩的情况下,map中间结果能缩小将近10倍,如果map的瓶颈在磁盘,那么job的性能提升将会非常可观。

当采用map中间结果压缩的情况下,用户还可以选择压缩时采用哪种压缩格式进行压缩,现在hadoop支持的压缩格式有:GzipCodec,LzoCodec,BZip2Codec,LzmaCodec等压缩格式。通常来说,想要达到比较平衡的cpu和磁盘压缩比,LzoCodec比较适合。但也要取决于job的具体情况。用户若想要自行选择中间结果的压缩算法,可以设置配置参数:mapred.map.output.compression.codec=org.apache.hadoop.io.compress.DefaultCodec或者其他用户自行选择的压缩方式。

1.2 Map side相关参数调优

选项 类型 默认值 描述
io.sort.mb int 100 缓存map中间结果的buffer大小(in MB)
io.sort.record.percent float 0.05 io.sort.mb中用来保存map output记录边界的百分比,其他缓存用来保存数据
io.sort.spill.percent float 0.80 map开始做spill操作的阈值
io.sort.factor int 10 做merge操作时同时操作的stream数上限。
min.num.spill.for.combine int 3 combiner函数运行的最小spill数
mapred.compress.map.output boolean false map中间结果是否采用压缩
mapred.map.output.compression.codec class name org.apache.hadoop.io.

compress.DefaultCodec

map中间结果的压缩格式

2 Reduce side tuning参数

2.1 ReduceTask运行内部原理

reduce的运行是分成三个阶段的。分别为copy->sort->reduce。由于job的每一个map都会根据reduce(n)数将数据分成map 输出结果分成n个partition,所以map的中间结果中是有可能包含每一个reduce需要处理的部分数据的。所以,为了优化reduce的执行时间,hadoop中是等job的第一个map结束后,所有的reduce就开始尝试从完成的map中下载该reduce对应的partition部分数据。这个过程就是通常所说的shuffle,也就是copy过程。

Reduce task在做shuffle时,实际上就是从不同的已经完成的map上去下载属于自己这个reduce的部分数据,由于map通常有许多个,所以对一个reduce来说,下载也可以是并行的从多个map下载,这个并行度是可以调整的,调整参数为:mapred.reduce.parallel.copies(default 5)。默认情况下,每个只会有5个并行的下载线程在从map下数据,如果一个时间段内job完成的map有100个或者更多,那么reduce也最多只能同时下载5个map的数据,所以这个参数比较适合map很多并且完成的比较快的job的情况下调大,有利于reduce更快的获取属于自己部分的数据。

reduce的每一个下载线程在下载某个map数据的时候,有可能因为那个map中间结果所在机器发生错误,或者中间结果的文件丢失,或者网络瞬断等等情况,这样reduce的下载就有可能失败,所以reduce的下载线程并不会无休止的等待下去,当一定时间后下载仍然失败,那么下载线程就会放弃这次下载,并在随后尝试从另外的地方下载(因为这段时间map可能重跑)。所以reduce下载线程的这个最大的下载时间段是可以调整的,调整参数为:mapred.reduce.copy.backoff(default 300秒)。如果集群环境的网络本身是瓶颈,那么用户可以通过调大这个参数来避免reduce下载线程被误判为失败的情况。不过在网络环境比较好的情况下,没有必要调整。通常来说专业的集群网络不应该有太大问题,所以这个参数需要调整的情况不多。

Reduce将map结果下载到本地时,同样也是需要进行merge的,所以io.sort.factor的配置选项同样会影响reduce进行merge时的行为,该参数的详细介绍上文已经提到,当发现reduce在shuffle阶段iowait非常的高的时候,就有可能通过调大这个参数来加大一次merge时的并发吞吐,优化reduce效率。

Reduce在shuffle阶段对下载来的map数据,并不是立刻就写入磁盘的,而是会先缓存在内存中,然后当使用内存达到一定量的时候才刷入磁盘。这个内存大小的控制就不像map一样可以通过io.sort.mb来设定了,而是通过另外一个参数来设置:mapred.job.shuffle.input.buffer.percent(default 0.7),这个参数其实是一个百分比,意思是说,shuffile在reduce内存中的数据最多使用内存量为:0.7 × maxHeap of reduce task。也就是说,如果该reduce task的最大heap使用量(通常通过mapred.child.java.opts来设置,比如设置为-Xmx1024m)的一定比例用来缓存数据。默认情况下,reduce会使用其heapsize的70%来在内存中缓存数据。如果reduce的heap由于业务原因调整的比较大,相应的缓存大小也会变大,这也是为什么reduce用来做缓存的参数是一个百分比,而不是一个固定的值了。

假设mapred.job.shuffle.input.buffer.percent为0.7,reduce task的max heapsize为1G,那么用来做下载数据缓存的内存就为大概700MB左右,这700M的内存,跟map端一样,也不是要等到全部写满才会往磁盘刷的,而是当这700M中被使用到了一定的限度(通常是一个百分比),就会开始往磁盘刷。这个限度阈值也是可以通过job参数来设定的,设定参数为:mapred.job.shuffle.merge.percent(default 0.66)。如果下载速度很快,很容易就把内存缓存撑大,那么调整一下这个参数有可能会对reduce的性能有所帮助。

当reduce将所有的map上对应自己partition的数据下载完成后,就会开始真正的reduce计算阶段(中间有个sort阶段通常时间非常短,几秒钟就完成了,因为整个下载阶段就已经是边下载边sort,然后边merge的)。当reduce task真正进入reduce函数的计算阶段的时候,有一个参数也是可以调整reduce的计算行为。也就是:mapred.job.reduce.input.buffer.percent(default 0.0)。由于reduce计算时肯定也是需要消耗内存的,而在读取reduce需要的数据时,同样是需要内存作为buffer,这个参数是控制,需要多少的内存百分比来作为reduce读已经sort好的数据的buffer百分比。默认情况下为0,也就是说,默认情况下,reduce是全部从磁盘开始读处理数据。如果这个参数大于0,那么就会有一定量的数据被缓存在内存并输送给reduce,当reduce计算逻辑消耗内存很小时,可以分一部分内存用来缓存数据,反正reduce的内存闲着也是闲着。

2.2 Reduce side相关参数调优

选项 类型 默认值 描述
mapred.reduce.parallel.copies int 5 每个reduce并行下载map结果的最大线程数
mapred.reduce.copy.backoff int 300 reduce下载线程最大等待时间(in sec)
io.sort.factor int 10 同上
mapred.job.shuffle.input.buffer.percent float 0.7 用来缓存shuffle数据的reduce task heap百分比
mapred.job.shuffle.merge.percent float 0.66 缓存的内存中多少百分比后开始做merge操作
mapred.job.reduce.input.buffer.percent float 0.0 sort完成后reduce计算阶段用来缓存数据的百分比

24 Responses

  1. adebug 说道:

    说得很全面很深入啊,学习了

    reduce阶段还有一个可以优化的点,reduce在拖数据的时候,从一台机器一次性把多个map结果拖过来,而不是一次拖一点。
    这个很久之前在30台的集群上测试过,由于一开始跑不满网络,只能到10M左右,每次shuffle都特别慢,因此改进了这个功能,直接单机网络打满,可见这里也是有可改进的地方的,但是对于大规模的环境下时,由于瓶颈已经不在网络上,往往是在io下,因此这个feature不是特别有效。

  2. rqetqr 说道:

    好文章,赞

  3. gaoyang 说道:

    好文,收藏了。

  4. Shen 说道:

    >>所以,为了优化reduce的执行时间,hadoop中是等job的第一个map结束后,所有的reduce就开始尝试从完成的map中下载该reduce对应的partition部分数据。

    這句話不對,實際上來說Reduce階段是等Job已完成5%的Map tasks數量之後才開始啟動Reduce,此參數為:mapred.reduce.slowstart.completed.maps

  5. guili 说道:

    @Shen 我们的hadoop版本里没有引入slowstart这个功能,所以…… ,呵呵

  6. Shen 说道:

    哦哦! 是採用0.19嗎?還是針對需求進行客製化的改寫?

  7. guili 说道:

    @Shen 从08年开始基于0.19.1版本,做了无数的需求定制化改进

  8. panfei 说道:

    感谢分享经验

  9. panfei 说道:

    能分享一下你们是怎么检测hadoop作业的瓶颈的方法?

  10. guili 说道:

    @panfei 这个说起来会很长,我们会在博客上持续更新分享我们的经验的~

  11. Raymond 说道:

    写的很细致,很清晰。Hadoop在淘宝一定被优化了很多东西,楼主是否有整理和分享的计划?

  12. guili 说道:

    @Raymond 多谢关注,我们会持续分享我们的经验,甚至开源我们的代码的。

  13. yfan 说道:

    好文,收获不少,你们非常open

  14. Roger 说道:

    学习了,收获很多。写得非常清晰,非常好。
    请问可以转载吗?

  15. guili 说道:

    @Roger 转载还请说明出处,谢谢~

  16. hadoop 说道:

    说的很好很全面,比definitive guide讲的还透彻,做performance tuning推荐一个工具ganglia,用于监控cluster的cpu,mem,network io等指标,逐一把硬件资源吃慢就tune到极限了;-)

  17. guili 说道:

    @楼上
    呵呵,多谢,我们会继续分享我们的经验,欢迎交流。
    另外,ganglia我们从使用hadoop的第一天就在使用,还做了很多改造。内部我们不叫ganglia,都称它为“肛裂”:)

  18. Denny Ye 说道:

    Shuffle过程中, Reduce端需要不断地对从map端拉过来的数据做merge。最好的情况是就在内存中做merge,完了后直接作为reduce的输入数据。

    开启这种MemToMem merge模式是将mapreduce.reduce.merge.memtomem.enabled设置为true。 当copy过来的数据太多时,超过一个内存限制,就会往磁盘做merge,我们应该最大限度地减少往磁盘merge,在这里设置mapreduce.reduce.shuffle.merge.percent,这个一个利用memoryLimit的比例,我给它设置为0.99,最大化地利用reduce端的内存来做merge。最后在做过一些内存间的merge后,你希望有多少内存来存储reduce的输入数据?mapreduce.reduce.input.buffer.percent,这个比例设置高些,这个数据设置的越大,在merge后期往磁盘merge数据就越少。我测试时设置是0.9。

    在我的测试例子中,reduce端的所有merge处理都是在内存中发生,没有往磁盘写一点数据。merge流程从最初的接近8分钟减少到2分钟,这2分钟基本就是在拉取数据,内存中merge是很快的。

    当然基于测试的数据是很理想化的,但上述的设置还是能最大化地减少reduce端对磁盘IO的操作。
    因为我测试的初衷是尽量让磁盘读写不要成为瓶颈。对于performance tuning的各种问题我们可以交流下,呵呵

  19. Denny Ye 说道:

    还有一个问题。我不清楚你的这些分析是基于Hadoop哪个版本的,但参数明显是挺老了,参数设置也很不灵活,很多时间没有将map与reduce的执行分开对待。只是个人感觉了

  20. guili 说道:

    @Denny Ye
    我们的hadoop版本是0.19.1,至于为什么到现在还用这个版本就不详细解释了,总之是历史原因。
    关于tunning的经验,可以给我发邮件交流:luoli523@gmail.com

  21. [...] 10.05pt;mso-layout-grid-align:none;text-autospace:none">1TB排序 MapReduce hadoop的源码分析 hadoop的源码分析 Hadoop MapReduce优化 Tuning Hadoop [...]

  22. babyyellow 说道:

    感谢分享!!!

  23. arkilis 说道:

    reduce的运行是分成三个阶段的。分别为copy->sort->reduce。

    ==========================================================
    请教一下,这个copy 是复制什么,从哪里复制到哪里?
    谢谢

  24. jijiannan 说道:

    @arkilis
    copy是指将map的输出拷贝到运行Reduce的节点上

from: http://www.tbdata.org/archives/1470

 
 
  • 标签:hadoop 
  • 发表评论:
    载入中。。。

     
     
     

    梦翔儿网站 梦飞翔的地方 http://www.dreamflier.net
    中华人民共和国信息产业部TCP/IP系统 备案序号:辽ICP备09000550号

    Powered by Oblog.