MapReduce: 提升MapReduce性能的七点建议【译】

做者是经过这个网站翻译过来的：

http://blog.cloudera.com/blog/2009/12/7-tips-for-improving-mapreduce-performance/   

    Cloudera提供给客户的服务内容之一就是调整和优化MapReduce job执行性能。MapReduce和HDFS组成一个复杂的分布式系统，而且它们运行着各式各样用户的代码，这样致使没有一个快速有效的规则来实现优化代码性能的目的。在我看来，调整cluster或job的运行更像一个医生对待病人同样，找出关键的“症状”，对于不一样的症状有不一样的诊断和处理方式。

        在医学领域，没有什么能够代替一位经验丰富的医生；在复杂的分布式系统上，这个道理依然正确—有经验的用户和操做者在面对不少常见问题上都会有“第六感”。我曾经为Cloudera不一样行业的客户解决过问题，他们面对的工做量、数据集和cluster硬件有很大区别，所以我在这方面积累了不少的经验，而且想把这些经验分享给诸位。

        在这篇blog里，我会高亮那些提升MapReduce性能的建议。前面的一些建议是面向整个cluster的，这可能会对cluster 操做者和开发者有帮助。后面一部分建议是为那些用Java编写MapReduce job的开发者而提出。在每个建议中，我列出一些“症状”或是“诊断测试”来讲明一些针对这些问题的改进措施，可能会对你有所帮助。

        请注意，这些建议中包含不少我以往从各类不一样场景下总结出来的直观经验。它们可能不太适用于你所面对的特殊的工做量、数据集或cluster，若是你想使用它，就须要测试使用前和使用后它在你的cluster环境中的表现。对于这些建议，我会展现一些对比性的数据，数据产生的环境是一个4个节点的cluster来运行40GB的Wordcount job。应用了我如下所提到的这些建议后，这个job中的每一个map task大概运行33秒，job总共执行了差很少8分30秒。

第一点  正确地配置你的Cluster
诊断结果/症状：
1. Linux top命令的结果显示slave节点在全部map和reduce slot都有task运行时依然很空闲。
2. top命令显示内核的进程，如RAID(mdX_raid*)或pdflush占去大量的CPU时间。
3. Linux的平均负载一般是系统CPU数量的2倍。
4. 即便系统正在运行job，Linux平均负载老是保持在系统CPU数量的一半的状态。
5. 一些节点上的swap利用率超过几MB

    优化你的MapReduce性能的第一步是确保你整个cluster的配置文件被调整过。对于新手，请参考这里关于配置参数的一篇blog：配置参数。 除了这些配置参数 ，在你想修改job参数以期提升性能时，你应该参照下我这里的一些你应该注意的项：

1.  确保你正在DFS和MapReduce中使用的存储mount被设置了noatime选项。这项若是设置就不会启动对磁盘访问时间的记录，会显著提升IO的性能。

2. 避免在TaskTracker和DataNode的机器上执行RAID和LVM操做，这一般会下降性能

3. 在这两个参数mapred.local.dir和dfs.data.dir 配置的值应当是分布在各个磁盘上目录，这样能够充分利用节点的IO读写能力。运行 Linux sysstat包下的iostat -dx 5命令可让每一个磁盘都显示它的利用率。

4. 你应该有一个聪明的监控系统来监控磁盘设备的健康状态。MapReduce job的设计是可容忍磁盘失败，但磁盘的异常会致使一些task重复执行而使性能降低。若是你发如今某个TaskTracker被不少job中列入黑名单，那么它就可能有问题。

5. 使用像Ganglia这样的工具监控并绘出swap和网络的利用率图。若是你从监控的图看出机器正在使用swap内存，那么减小mapred.child.java.opts属性所表示的内存分配。

基准测试：
    很遗憾我不能为这个建议去生成一些测试数据，由于这须要构建整个cluster。若是你有相关的经验，请把你的建议及结果附到下面的留言区。

第二点  使用LZO压缩
诊断结果/症状：
1. 对 job的中间结果数据使用压缩是很好的想法。
2. MapReduce job的输出数据大小是不可忽略的。
3. 在job运行时，经过linux top 和 iostat命令能够看出slave节点的iowait利用率很高。

    几乎每一个Hadoop job均可以经过对map task输出的中间数据作LZO压缩得到较好的空间效益。尽管LZO压缩会增长一些CPU的负载，但在shuffle过程当中会减小磁盘IO的数据量，整体上老是能够节省时间的。

    当一个job须要输出大量数据时，应用LZO压缩能够提升输出端的输出性能。这是由于默认状况下每一个文件的输出都会保存3个幅本，1GB的输出文件你将要保存3GB的磁盘数据，当采用压缩后固然更能节省空间并提升性能。

    为了使LZO压缩有效，请设置参数mapred.compress.map.output值为true。

基准测试：
    在个人cluster里，Wordcount例子中不使用LZO压缩的话，job的运行时间只是稍微增长。但FILE_BYTES_WRITTEN计数器却从3.5GB增加到9.2GB，这表示压缩会减小62%的磁盘IO。在个人cluster里，每一个数据节点上磁盘数量对task数量的比例很高，但Wordcount job并无在整个cluster中共享，因此cluster中IO不是瓶颈，磁盘IO增加不会有什么大的问题。但对于磁盘因不少并发活动而受限的环境来讲，磁盘IO减小60%能够大幅提升job的执行速度。

第三点  调整map和reduce task的数量到合适的值

本身的经验，通常来讲，不可能跑一个job， 改变整个集群的hdfs block的大小。一般提交job时候设置参数。

   job.setMapOutputValueClass(IntWritable.class);
            job.setNumReduceTasks(1);
            //设置最小分片为512M
            FileInputFormat.setMinInputSplitSize(job, 1024*1024*512);
            FileInputFormat.addInputPath(job, new Path("/usr/keyword/input"));

 

诊断结果/症状：
1. 每一个map或reduce task的完成时间少于30到40秒。
2. 大型的job不能彻底利用cluster中全部空闲的slot。
3. 大多数map或reduce task被调度执行了，但有一到两个task还在准备状态，在其它task完成以后才单独执行

    调整job中map和reduce task的数量是一件很重要且经常被忽略的事情。下面是我在设置这些参数时的一些直观经验：

1. 若是每一个task的执行时间少于30到40秒，就减小task的数量。Task的建立与调度通常耗费几秒的时间，若是task完成的很快，咱们就是在浪费时间。同时，设置JVM重用也能够解决这个问题。

2. 若是一个job的输入数据大于1TB，咱们就增长block size到256或者512，这样能够减小task的数量。你可使用这个命令去修改已存在文件的block size: hadoop distcp -Ddfs.block.size=$[256*1024*1024] /path/to/inputdata  /path/to/inputdata-with/largeblocks。在执行完这个命令后，你就能够删除原始的输入文件了(/path/to/inputdata)。

3. 只要每一个task运行至少30到40秒，那么就增长map task的数量，增长到整个cluster上map slot总数的几倍。若是你的cluster中有100个map slot，那就避免运行一个有101个map task的job — 若是运行的话，前100个map同时执行，第101个task会在reduce执行以前单独运行。这个建议对于小型cluste和小型job是很重要的。

4. 不要调度太多的reduce task — 对于大多数job来讲，咱们推荐reduce task的数量应当等于或是略小于cluster中reduce slot的数量。

基准测试：
    为了让Wordcount job有不少的task运行，我设置了以下的参数：Dmapred.max.split.size=$[16*1024*1024]。之前默认会产生360个map task，如今就会有2640个。当完成这个设置以后，每一个task执行耗费9秒，而且在JobTracker的Cluster Summar视图中能够观看到，正在运行的map task数量在0到24之间浮动。job在17分52秒以后结束，比原来的执行要慢两倍多。

第四点  为job添加一个Combiner
诊断结果/症状：
1. job在执行分类的聚合时，REDUCE_INPUT_GROUPS计数器远小于REDUCE_INPUT_RECORDS计数器。
2. job执行一个大的shuffle任务(例如，map的输出数据每一个节点就是好几个GB)。
3. 从job计数器中看出，SPILLED_RECORDS远大于MAP_OUTPUT_RECORDS。

    若是你的算法涉及到一些分类的聚合，那么你就可使用Combiner来完成数据到达reduce端以前的初始聚合工做。MapReduce框架很明智地运用Combiner来减小写入磁盘以及经过网络传输到reduce端的数据量。

基准测试：
    我删去Wordcount例子中对setCombinerClass方法的调用。仅这个修改就让map task的平均运行时间由33秒增加到48秒，shuffle的数据量也从1GB提升到1.4GB。整个job的运行时间由原来的8分30秒变成15分42秒，差很少慢了两倍。此次测试过程当中开启了map输出结果的压缩功能，若是没有开启这个压缩功能的话，那么Combiner的影响就会变得更加明显。

第五点  为你的数据使用最合适和简洁的Writable类型
诊断/症状：
1. Text 对象在非文本或混合数据中使用。
2. 大部分的输出值很小的时候使用IntWritable 或 LongWritable对象。

    当一个开发者是初次编写MapReduce，或是从开发Hadoop Streaming转到Java MapReduce，他们会常常在没必要要的时候使用Text 对象。尽管Text对象使用起来很方便，但它在由数值转换到文本或是由UTF8字符串转换到文本时都是低效的，且会消耗大量的CPU时间。当处理那些非文本的数据时，可使用二进制的Writable类型，如IntWritable， FloatWritable等。

    除了避免文件转换的消耗外，二进制Writable类型做为中间结果时会占用更少的空间。当磁盘IO和网络传输成为大型job所遇到的瓶颈时，减小些中间结果的大小能够得到更好的性能。在处理整形数值时，有时使用VIntWritable或VLongWritable类型可能会更快些—这些实现了变长整形编码的类型在序列化小数值时会更节省空间。例如，整数4会被序列化成单字节，而整数10000会被序列化成两个字节。这些变长类型用在统计等任务时更加有效，在这些任务中咱们只要确保大部分的记录都是一个很小的值，这样值就能够匹配一或两个字节。

    若是Hadoop自带的Writable类型不能知足你的需求，你能够开发本身的Writable类型。这应该是挺简单的，可能会在处理文本方面更快些。若是你编写了本身的Writable类型，请务必提供一个RawComparator类—你能够之内置的Writable类型作为例子。

基准测试：
    对于Wordcount例子，我修改了它在map计数时的中间变量，由IntWritable改成Text。而且在reduce统计最终和时使用Integer.parseString(value.toString)来转换出真正的数值。这个版本比原始版本要慢近10%—整个job完成差很少超过9分钟，且每一个map task要运行36秒，比以前的33秒要慢。尽可能看起来整形转换仍是挺快的，但这不说明什么状况。在正常状况下，我曾经看到过选用合适的Writable类型能够有2到3倍的性能提高的例子。

第六点  重用Writable类型
诊断/症状：
1. 在mapred.child.java.opts参数上增长-verbose:gc -XX:+PriintGCDetails，而后查看一些task的日志。若是垃圾回收频繁工做且消耗一些时间，你须要注意那些无用的对象。
2. 在你的代码中搜索"new Text" 或"new IntWritable"。若是它们出如今一个内部循环或是map/reduce方法的内部时，这条建议可能会颇有用。
3. 这条建议在task内存受限的状况下特别有用。

    不少MapReduce用户常犯的一个错误是，在一个map/reduce方法中为每一个输出都建立Writable对象。例如，你的Wordcout mapper方法可能这样写：

Java代码

 

1. public void map(...) {  
2.       …  
3.       for (String word : words) {  
4.               output.collect(new Text(word), new IntWritable(1));  
5.       }  
6. }  

    这样会致使程序分配出成千上万个短周期的对象。Java垃圾收集器就要为此作不少的工做。更有效的写法是：

Java代码

 

1. class MyMapper … {  
2.    Text wordText = new Text();  
3.    IntWritable one = new IntWritable(1);  
4.    public void map(...) {  
5.          for (String word: words) {  
6.                wordText.set(word);  
7.                output.collect(wordText, one);  
8.           }  
9.        }  
10. }  

基准测试：
    当我以上面的描述修改了Wordcount例子后，起初我发现job运行时与修改以前没有任何不一样。这是由于在个人cluster中默认为每一个task都分配一个1GB的堆大小 ，因此垃圾回收机制没有启动。当我从新设置参数，为每一个task只分配200MB的堆时，没有重用Writable对象的这个版本执行出现了很严重的减缓 —job的执行时间由之前的大概8分30秒变成如今的超过17分钟。原始的那个重用Writable的版本，在设置更小的堆时仍是保持相同的执行速度。所以重用Writable是一个很简单的问题修正，我推荐你们老是这样作。它可能不会在每一个job的执行中得到很好的性能，但当你的task有内存限制时就会有至关大的区别。

第七点  使用简易的剖析方式查看task的运行
    这是我在查看MapReduce job性能问题时经常使用的一个小技巧。那些不但愿这些作的人就会反对说这样是行不通的，可是事实是摆在面前。

    为了实现简易的剖析，能够当job中一些task运行很慢时，用ssh工具链接上task所在的那台task tracker机器。执行5到10次这个简单的命令 sudo killall -QUIT java(每次执行间隔几秒)。别担忧，不要被命令的名字吓着，它不会致使任何东西退出。而后使用JobTracker的界面跳转到那台机器上某个task的stdout 文件上，或者查看正在运行的机器上/var/log/hadoop/userlogs/目录中那个task的stdout文件。你就能够看到当你执行那段命令时，命令发送到JVM的SIGQUIT信号而产生的栈追踪信息的dump文件。([译]在JobTracker的界面上有Cluster Summary的表格，进入Nodes连接，选中你执行上面命令的server，在界面的最下方有Local Logs,点击LOG进入，而后选择userlogs目录，这里能够看到以server执行过的jobID命名的几个目录，无论进入哪一个目录均可以看到不少task的列表，每一个task的log中有个stdout文件，若是这个文件不为空，那么这个文件就是做者所说的栈信息文件)

    解析处理这个输出文件须要一点点以经验，这里我介绍下平时是怎样处理的：
对于栈信息中的每一个线程，很快地查找你的java包的名字(假如是com.mycompany.mrjobs)。若是你当前线程的栈信息中没有找到任何与你的代码有关的信息，那么跳到另外的线程再看。

    若是你在某些栈信息中看到你查找的代码，很快地查阅并大概记下它在作什么事。假如你看到一些与NumberFormat相关的信息，那么此时你须要记下它，暂时不须要关注它是代码的哪些行。

    转到日志中的下一个dump，而后也花一些时间作相似的事情而后记下些你关注的内容。

    在查阅了4到5个栈信息后，你可能会意识到在每次查阅时都会有一些似曾相识的东西。若是这些你意识到的问题是阻碍你的程序变快的缘由，那么你可能就找到了程序真正的问题。假如你取到10个线程的栈信息，而后从5个里面看到过NumberFormat相似的信息，那么可能意味着你将50%的CPU浪费在数据格式转换的事情上了。

    固然，这没有你使用真正的分析程序那么科学。但我发现这是一种有效的方法，能够在不须要引入其它工具的时候发现那些明显的CPU瓶颈。更重要的是，这是一种让你会变的更强的技术，你会在实践中知道一个正常的和有问题的dump是啥样子。

    经过这项技术我发现了一些一般出如今性能调优方面的误解，列出在下面。
1. NumberFormat 至关慢，尽可能避免使用它。
2. String.split—无论是编码或是解码UTF8的字符串都是慢的超出你的想像— 参照上面提到的建议，使用合适的Writable类型。
3. 使用StringBuffer.append来链接字符串

    上面只是一些提升MapReduce性能的建议。作基准测试的那些代码我放在了这里：performance blog code