hadoop JOB的性能优化实践

ty9919

　　使用了几个月的hadoopMR，对遇到过的性能问题做点笔记，这里只涉及job的性能优化，没有接触到
　　hadoop集群，操作系统，任务调度策略这些方面的问题。
　　hadoop MR在做大数据量分析时候有限的计算资源情况下只能不断的优化程序。
　　优化可以从两个方面进行：
　　1.hadoop配置
　　2.程序代码
　　程序代码包括的方面很多：job设计，算法，数据结构，代码编写。
hadoop配置优化
　　hadoop配置可分为mapp配置，reducer配置和hdfs配置。关于hadoop mapper和reducer阶段
　　处理流程和参数意义可以看这个帖子，说的比较详细hadoop mr 参数意义。
　　这里再补充几个配置：

　　dfs.block.size
　　这个配置项定义了在HDFS上每个block的大小，它的值是以字节为单位。
　　可以在配置文件hadoop-site.xml(Hadoop 0.20 以前版本)定义，
　　也可以在JobConf里定义。hdfs中block size定义是以文件为粒度的。
　　hadoop的mapper数基本由输入文件的block数决定，如果输入的block
　　size不够大，导致mapper处理时间很短(不到一分钟)，大量这样的mapper
　　会严重降低计算性能。但是如果输入文件都是小文件，就算blocksize再大，每个
　　文件也会占一个block，这时候要通过合并小文件来减少mapper数，设置blocksize
　　是没用的。命令行设置块大小可以加参数，0.20以后的用
　　hadoop fs -D dfs.block.size=134217728 -put local_name remote_location
　　之前的可以用fs.local.block.size 参数
　　除了blocksize hadoop的inputformat也提供了在block的基础上更细粒度控制mapper
　　输入块大小，比如当前输入块128M，设置了最大分割size为64，则原先一个块被切分
　　成两个spliter了，也就产生了两个mapper。用这种方法可以有效增加mapper数，但对减少
　　mapper数好像没用。
　　FileInputFormat.setMaxInputSplitSize(job, size)
　　FileInputFormat.setMinInputSplitSize(job, size)
　　mapred.min.split.size这个参数也可以起到同样效果
　　mapred.map.tasks.speculative.execution 和 
　　mapred.reduce.tasks.speculative.execution
　　这两个选项是设置推测执行的任务，当所有task都开始运行之后，Job Tracker会统计所有任务的平均进度，
　　如果某个task所在的task node机器配置比较低或者CPU load很高（原因很多），导致任务执行比总体任务的平均执行要慢，
　　此时Job Tracker会启动一个新的任务（duplicate task），这个新任务就是推测任务，原有任务和新任务哪个先执行完就把另外一个kill掉，
　　这也是我们经常在Job Tracker页面看到任务执行成功，但是总有些任务被kill，就是这个原因。推测任务也是要占用计算资源，
　　因此计算资源紧张，任务执行本身很耗资源情况下可以考虑设置成false，禁止执行。
　　io.sort.mb
　　以MB为单位，默认100M，通常来看，这个值太小了，这个选项定义了map输出结果在内存占用buffer的大小，当buffer达到一定阈值，
　　会启动一个后台线程来对buffer的内容进行排序，然后写入本地磁盘(一个spill文件)。可以观察hadoop的日志，如果spill次数比较多说明
　　这个缓存大小设置太低，特别是那种mapper中处理数据会增多的逻辑尤其可以关注下。
　　根据map输出数据量的大小，可以适当的调整buffer的大小，注意是适当的调整，不是越大越好，假设内存无限大，io.sort.mb=1024(1G),
　　和io.sort.mb=300 (300M)，前者未必比后者快，因为1G的数据排序一次和排序3次，每次300MB，一定是后者快(分而治之的思想)。
　　io.sort.spill.percent
　　这个值就是上述buffer的阈值，默认是0.8，既80%，当buffer中的数据达到这个阈值，后台线程会起来对buffer中已有的数据进行排序，
　　然后写入磁盘，此时map输出的数据继续往剩余的20% buffer写数据，如果buffer的剩余20%写满，排序还没结束，map task被block等待。
　　如果你确认map输出的数据基本有序(很少见)，排序时间很短，可以将这个阈值适当调高，更理想的，如果你的map输出是有序的数据（基本不可能吧？），
　　那么可以把buffer设的更大，阈值设置为1.
　　Io.sort.factor
　　同时打开磁盘spill进行并行合并的文件数，默认是10。
　　当一个map task执行完之后，本地磁盘上(mapred.local.dir)有若干个spill文件，map task最后做的一件事就是执行merge sort，
　　把这些spill文件合成一个文件（partition），有时候我们会自定义partition函数，就是在这个时候被调用的。
　　执行merge sort的时候，每次同时打开多少个spill文件，就是由io.sort.factor决定的。打开的文件越多，不一定merge sort就越快，所以也要根据数据情况适当的调整。
　　补充：merge排序的结果是两个文件，一个是index，另一个是数据文件，index文件记录了每个不同的key在数据文件中的偏移量（这就是partition）

代码优化
　　有空再写
　　各种配置
　　Mapper端配置
　　

　　1.Map逻辑处理后数据被展开，写磁盘次数剧增，可以观察日志中的spill次数，调整各个参数
　　

　　2.中间结果能不展开就不展开，尽量缩小Mapper和reducer之间的数据传递
　　

　　3.distribute cache中加载的数据能不用hashmap就尽量不要用，hashmap会使得内存占用量是原数据的5-10倍，其中
　　引用占了大量空间
　　

　　4.distribute cache中加载的数据要尽可能简单，如果有复杂的处理逻辑可以单独开辟Mapper Reducer进行一轮处理，
　　避免每次mapper都要处理一遍，尽可能减少distribute cache的数据量
　　

　　5.观察GC的情况，有时候是因为内存占用量高，频繁GC，严重影响处理速度
　　

　　6.当逻辑本身很简单，但是处理速度很慢时候首先要怀疑Mapper和Reducer之间传输数据量过大，其次是GC情况
　　

　　7.适当控制mapper的数量，特别是有distribute cache的场景