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相关参数调优
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相关参数调优
本文转载自：http://www.tbdata.org/archives/1470
此文章是淘宝的数据团队博客,下面再说些其他方面的优化
hadoop优化相关: 1:对操作系统进行参数调优 (1):打开文件描述符和网络连接参数上限(具体操作内容:使用ulimit命令讲允许同时打开的文件描述符数据上限增大至一个合适的值,同时调整内核参数net.core.somaxconn) (2):关闭swap分区(具体操作内容是/etc/stsctl.conf中得vm.vm.swappiness参数) (3):设置合理的预读取缓冲区大小(具体操作内容:使用linux命令blockdev设置预读取缓冲区的大小) (4):文件系统的选择和配置 2:JVM参数的优化 3:通过hadoop的参数进行调优 (1):设置合理的槽位数目(具体配置 mapred.tasktracker.map.tasks.maximum | mapred.tasktracker.reduce.tasks.maximum | mapreduce.tasktracker.map.tasks.maximum | mapreduce.tasktracker.reduce.tasks.maximum) (2):调整心跳间隔,对于300台以下的集群 可以把心跳设置成300毫秒(默认是3秒),mapreduce.jobtracker.hearbeat.interval.min | mapred.hearbeats.in.second | mapreduce.jobtracker.heartbeats.scaling.factor (3):启用外心跳,为了减少任务分配延迟(比如我们的任务心跳设置为10秒钟,当有一个任务挂掉了之后,他就不能马上通知jobtracker), 所以hadoop引入了外心跳,外心跳是任务运行结束或者任务运行失败的时候触发的,能够在出现空闲资源时第一时间通知jobtracker,以便他能够迅速为空闲资源分配新的任务 外心跳的配置参数是 mapreduce.tasktracker.outofband.hearbeat (4):磁盘快的配置. map task会把中间结果放到本地磁盘中,所以对于I/O密集的任务来说这部分数据会对本地磁盘造成很大的压力,我们可以配置多块可用磁盘,hadoop将采用轮训的方式将不同的maptask的中间结果写到磁盘上 maptask中间结果的配置参数是mapred.local.dir | mapreduce.cluster.local.dir (5):配置RPC Handler的数量,jobracker需要冰法处理来自各个tasktracker的RPC请求,我们可以根据集群规模和服务器并发处理的情况调整RPC Handler的数目,以使jobtracker的服务能力最佳 配置参数是 mapred.job.tracker.handler.count | mapreduce.jobtracker.handler.count (默认是10) (6):配置HTTP线程数. 在shuffle阶段,reduce task 通过http请求从各个tasktracker上读取map task中间结果,而每个tasktracker通过jetty server处理这些http请求,所以可以适当配置调整jetty server的工作线程数 配置参数是 tasktracker.http.thread | mapreduce.tasktracker.http.threads (默认是40) (7):如果我们在运行作业的过程中发现某些机器被频繁地添加到黑名单里面,我们可以把此功能关闭 (8):使用合理调度器 (9):使用合适的压缩算法,在hadoop里面支持的压缩格式是: gzip,zip,bzip2,LZO,Snappy,LZO和Snappy的呀搜比和压缩效率都很优秀,Snappy是谷歌的开源数据压缩哭,他已经内置在hadoop1.0之后的版本,LZO得自己去编译 (10):开启预读机制. 预读机制可以有效提高磁盘I/O的读性能,目前标准版的apache hadoop不支持此功能,但是在cdh中是支持的 配置参数是: mapred.tasktracker.shuffle.fadvise=true (是否启用shuffle预读取机制) mapred.tasktracker.shuffle.readahead.bytes=4MB (shuffle预读取缓冲区大小) mapreduce.ifile.readahead = true (是否启用ifile预读取机制) mapreduce.ifile.readahead.bytes = 4MB (IFile预读取缓冲区大小) (11):启用推测执行机制 (12):map task调优: 合理调整io.sort.record.percent值,可减少中间文件数据,提高任务执行效率. (map task的输出结果将被暂时存放到一个环形缓冲区中,这个缓冲区的大小由参数"io.sort.mb"指定,单位MB,默认是100MB, 该缓冲区主要由两部分组成,索引和实际数据,默认情况下,索引占整个buffer的比例为io.sort.record.percent,默认是5%, 剩余空间存放数据,仅当满足以下任意一个条件时才会触发一次flush,生成一个临时文件,索引或者数据空间使用率达到比例为 io.sort.spill.percent的80%) 所以具体调优参数如下: io.sort.mb | io.sort.record.percent | io.sort.spill.percent (13):reduce task调优 reduce task会启动多个拷贝线程从每个map task上读取相应的中间结果,参数是"mapred.reduce.parallel.copies"(默认是5) 原理是这样的-->对于每个待拷贝的文件,如果文件小于一定的阀值A,则将其放入到内存中,否则已文件的形式存放到磁盘上, 如果内存中文件满足一定条件D,则会将这些数据写入磁盘中,而当磁盘上文件数目达到io.sort.factor(默认是10)时, 所以如果中间结果非常大,可以适当地调节这个参数的值 (14):跳过坏记录 看具体参数说明,=号后面是默认值 mapred.skip.attempts.to.start.skipping=2 当任务失败次数达到该值时,才会进入到skip mode,即启用跳过坏记录gongnneg mapred.skip.map.max,skip.records=0 用户可通过该参数设置最多运行跳过的记录数目 mapred.skip.reduce.max.skip.groups=0 用户可通过设置该参数设置Reduce Task最多允许跳过的记录数目 mapred.skip.out.dir =${mapred.output.dir}/logs/ 检测出得坏记录存放到目录里面(一般为HDFS路径),hadoop将坏记录保存起来以便于用户调试和跟踪 (14):使用JVM重用 : mapred.job.reuse.jvm.aum.tasks | mapreduce.job.jvm.num.tasks = -1 4:从用户角度来优化 (1)设置combiner. 在应用中尽量使用combiner可以有效地提高效率 (2)选择合适的writable (3)设置合理的reduce数 (4)合理使用DistributedCache(建议如果需要一个外部文件引入的时候,事先把他上传到hdfs上,这样效率高,因为这样节省了客户端上传文件的时间,并且还隐含地告诉DistributedCache, 请将文件下载到各节点的public共享目录下) (5)合理控制Reduce Task的启动时机 ,因为在执行job的时候,reduce task晚于map task启动,所以合理控制reduce task启动时机不仅可以加快作业的运行速度 还可以提高资源利用率,如果reduce task启动过早,则可能由于reduce task长时间占用reduce slot资源造成slot hoarding现象,而且还会降低资源利用率 反之则导致reduce task获取资源延迟,增加了作业的运行时间.hadoop配置reduce task启动时机的参数是 mapred.reduce.slowstart.completed.maps | mapreduce.job.reduce.slowstart.completed.maps (默认值是0.05,也就是map task完成数目达到5%时,开始启动reduce task)
[Hive]从一个经典案例看优化mapred.map.tasks的重要性
我所在公司所使用的生产Hive环境的几个参数配置如下：
dfs.block.size=268435456
hive.merge.mapredfiles=true
hive.merge.mapfiles=true
hive.merge.size.per.task=256000000
mapred.map.tasks=2
因为合并小文件默认为true，而dfs.block.size与hive.merge.size.per.task的搭配使得合并后的绝大部分文件都在300MB左右。
CASE 1：
现在我们假设有3个300MB大小的文件，那么goalsize = min(900MB/2,256MB) = 256MB (具体如何计算map数请参见http://blog.sina.com.cn/s/blog_6ff05a2c010178qd.html)
所以整个JOB会有6个map，其中3个map分别处理256MB的数据，还有3个map分别处理44MB的数据。
这时候木桶效应就来了，整个JOB的map阶段的执行时间不是看最短的1个map的执行时间，而是看最长的1个map的执行时间。所以，虽然有3个map分别只处理44MB的数据，可以很快跑完，但它们还是要等待另外3个处理256MB的map。显然，处理256MB的3个map拖了整个JOB的后腿。
CASE 2：
如果我们把mapred.map.tasks设置成6，再来看一下有什么变化：
goalsize = min(900MB/6,256MB) = 150MB
整个JOB同样会分配6个map来处理，每个map处理150MB的数据，非常均匀，谁都不会拖后腿，最合理地分配了资源，执行时间大约为CASE 1的59%(150/256)
案例分析：
虽然mapred.map.tasks从2调整到了6，但是CASE 2并没有比CASE 1多用map资源，同样都是使用6个map。而CASE 2的执行时间约为CASE 1执行时间的59%。
从这个案例可以看出，对mapred.map.tasks进行自动化的优化设置其实是可以很明显地提高作业执行效率的。