`
langyu
  • 浏览: 883144 次
  • 性别: Icon_minigender_1
  • 来自: 杭州
社区版块
存档分类
最新评论

MapReduce: 提高MapReduce性能的七点建议[译]

阅读更多

        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.dirdfs.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的数量到合适的值
诊断结果/症状:
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方法可能这样写:

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


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

class MyMapper … {
   Text wordText = new Text();
   IntWritable one = new IntWritable(1);
   public void map(...) {
         for (String word: words) {
               wordText.set(word);
               output.collect(wordText, one);
          }
       }
}

基准测试:
    当我以上面的描述修改了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

原文:  7 Tips for Improving MapReduce Performance



3
0
分享到:
评论

相关推荐

Global site tag (gtag.js) - Google Analytics