Apache Hadoop最佳实践和反模式解析

　　Apache Hadoop是一个用来构建大规模共享存储和计算设施的软件。Hadoop集群已经应用在多种研究和开发项目中，并且，Yahoo!, EBay, Facebook, LinkedIn, Twitter等公司，越来越多的的把它应用在生产环境中。 这些已有的经验是技术和投入的结晶，在许多情况下至关重要。因此，适当的使用Hadoop集群可以保证我们的投入能够获得最佳回报。

　　这篇博文简单总结了一些Hadoop应用的最佳实践。实际上，类似于设计模式，我们引进一个网格模式的的概念，来提供一个通用且可复用的针对运行在网格上的应用的解决方案。

　　这篇博文列举了表现良好的应用的特点并且提供了正确使用Hadoop框架的各种特性和功能的指导。这些特点很大程度上这由其本身特点而定，阅读这篇文档的一个好方法是从本质上理解应用，这些最佳实践在Hadoop的多租户环境下卓有成效，而且不会与框架本身的大多数原则和限制产生矛盾。

　　博文还强调了一些Hadoop应用的反模式。

概述

　　Hadoop上的数据处理应用一般使用Map-Reduce模型。

　　一个Map-Reduce作业通常会把输入的数据集拆分成许多独立的数据段，按照完全并行的方式一个map任务处理一段。框架把map的输出排序，然后作为reduce的输入。通常输入和输出都储存在文件系统。框架负责调度、监控任务的执行以及重启失败的任务。

　　Map-Reduce应用可以指定输入输出的位置，并提供了map与reduce功能的实现，体现在Hadoop中是Mapper 和Reducer.这些只是作业配置的一部分参数。Hadoop客户端提交作业（jar或者其他可执行的程序）和配置给JobTracker，而JobTracker负责把程序和配置分发到各个slave，调度和监控任务的执行，并返回状态信息给客户端。

　　Map/Reduce框架的处理是基于<key, value>这样的键值对，也就是说，框架吧输入数据视作一系列<key, value>键值对集合，然后产出另一些键值对作为输出。

　　这是 Map-Reduce应用的典型数据流

　　绝大多数在网格上运行的Map-Reduce应用都不会直接实现较低层次的Map-Reduce接口，而是借助于较高抽象层次的语言，例如Pig。

　　Oozie 是一个非常好的网格上的工作流管理和调度方案。Oozie 支持多种应用接口 (Hadoop Map-Reduce, Pig, Hadoop Streaming, Hadoop Pipes, 等等.) 并且支持基于时间或数据可用性的调度。#p#

网格模式

　　这一部分是关于网格上运行的Map-Reduce应用的最佳实践

　　输入

　　Hadoop Map-Reduce 为处理海量数据而设计。maps过程以一种高度并行的方式来处理数据, 通常一个map至少处理一个HDFS block，一般是128M。

• 默认情况下，每个map最多处理一个HDFS 文件。 这意味着假如应用需要处理大量的文件，最好一个map能够处理多个。可以通过一种特定的输入格式来达成这个目的，就是MultiFileInputFormat。即使对于那些只处理很少小文件的应用，每个map处理多个文件的效率也更高。
• 假如应用需要处理的数据量非常大，即使文件尺寸很大，每个map处理128M以上的数据也会更有效率。

　　网格模式: 合并小文件以减少map数量，在处理大数据集的时候用比较大的HDFS 块大小。

　　Maps

　　maps的数量通常取决于输入大小, 也即输入文件的block数。 因此，假如你的输入数据有10TB，而block大小为128M，则需要82,000个map。

　　因为启动任务也需要时间，所以在一个较大的作业中，最好每个map任务的执行时间不要少于1分钟。

　　就像在上面“输入”部分所解释的，对于那种有大量小文件输入的的作业来说，一个map处理多个文件会更有效率。

　　如果应用处理的输入文件尺寸较大，每个map处理一个完整的HDFS block，数据段大一点更有效率。举个例子，让每个map处理更多数据，方法之一是让输入文件有更大的HDFS block尺寸，例如512M或者更多。

　　一个极端的例子是Map-Reduce开发团队用了大约66000个map来做PetaSort，也即66000个map要处理1PB数据，平均每个map 12.5G。

　　原则是大量运行时间很短的map会有损生产力。

　　网格模式：除非应用的map过程是CPU密集型，否则一个应用不应该有60000-70000个map。

　　当在map处理的block比较大的时候，确保有足够的内存作为排序缓冲区是非常重要的，这可以加速map端的排序过程。假如大多数的map输出都能在排序缓冲区中处理的话应用的性能会有极大的提升。这需要运行map过程的JVM具有更大的堆。记住反序列化输入的内存操作不同于磁盘操作；例如，Pig应用中的某些class将硬盘上的数据载入内存之后占用的空间会是其本来尺寸的3、4倍。在这种情况下，应用需要更大的JVM堆来让map的输入和输出数据能够保留在内存中。

　　网格模式：确保map的大小，使得所有的map输出可以在排序缓冲区中通过一次排序来完成操作。

　　合适的map数量有以下好处：

• 减少了调度的负担；更少的map意味着任务调度更简单，集群中可用的空闲槽更多。
• 有足够的内存将map输出容纳在排序缓存中，这使map端更有效率；
• 减少了需要shuffle map输出的寻址次数，每个map产生的输出可用于每一个reduce，因此寻址数就是map个数乘以reduce个数；
• 每个shuffled的片段更大，这减少了建立连接的相对开销，所谓相对开销是指相对于在网络中传输数据的过程。
• 这使reduce端合并map输出的过程更高效，因为合并的次数更少，因为需要合并的文件段更少了。

　　上述指南需要注意，一个map处理太多的数据不利于失败转移，因为单个map失败可能会造成应用的延迟。

　　Combiner

　　适当的使用Combiner可以优化map端的聚合。Combiner最主要的好处在于减少了shuffle过程从map端到reduce端的传输数据量。

　　Shuffle

　　适当的使用Combiner可以优化map端的聚合。Combiner最主要的好处在于减少了shuffle过程从map端到reduce端的传输数据量。

　　Combiner 也有一个性能损失点，因为它需要一次额外的对于map输出的序列化/反序列化过程。不能通过聚合将map端的输出减少到20-30%的话就不适用combiner。可以用 combiner input/output records counters（译者注：这是一个hadoop mapreduce 的counter名称，所以采用了原名未翻译）来衡量Combiner的效率。

　　网格模式：Combiners可以减少shuffle阶段的网络流量。但是，要保证Combiner 的聚合是确实有效的。

Reduces

　　reduces的性能很大程度上受shuffle的性能所影响。

　　应用配置的reduces数量是一个决定性的因素。

　　太多或者太少的reduce都不利于发挥最佳性能:

• 太少的reduce会使得reduce运行的节点处于过度负载状态，在极端情况下我们见过一个reduce要处理100g的数据。这对于失败恢复有着非常致命的负面影响，因为失败的reduce对作业的影响非常大。
• 太多的reduce对shuffle过程有不利影响。在极端情况下会导致作业的输出都是些小文件，这对NameNode不利，并且会影响接下来要处理这些小文件的mapreduce应用的性能。

　　网格模式：在大多数情况下，应用应该保证每个reduce处理1-2g数据，最多5-10g。

　　输出

　　我们需要记住一个重要的因素——应用的输出文件数取决于配置的reduce数。从我们上文中对reduce的讨论可知，reduce数的选择十分关键。

　　此外，还需要考虑其它一些因素:

• 考虑采用合适的压缩器（压缩速度vs性能）对输出进行压缩，提高HDFS的写入性能。
• 每个reduce不要输出多个文件，避免生成附属文件。我们一般用附属文件来记录统计信息，如果这些信息不多的话，可以使用计数器。
• 为输出文件选择合适的格式。对于下游消费者程序来说，用zlib/gzip/lzo等算法来对大量文本数据进行压缩往往事与愿违。因为zlib/gzip/lzo文件是不能分割的，只能整个进行处理。这会引起恶劣的负载均衡和故障恢复问题。作为改善，可以使用SequenceFile和TFile格式，它们不但是压缩的，而且是可以分割的。
• 如果每个输出文件都很大（若干GB），请考虑使用更大的输出块（dfs.block.size）。

　　网格模式: 应该确保应用的输出是数量不多的大文件，每个文件跨越多个HDFS块，而且经过适当的压缩。

　　分布式缓存（Distributed Cache）

　　分布式缓存可以高效的分发与具体应用相关的较大尺寸的只读文件。这是Map/Reduce框架提供的机制，用于暂时存储与特定应用相关的文件（如text, archives, jars等）。

　　这个框架会在slave节点上执行任务之前将必要的文件拷贝到该节点。它如此高效是因为在个作业中所需要的文件只会被复制一遍，还因为它能够缓存slave节点上的未归档文件。它也被作为基本软件分发机制用于map和reduce 任务。这种机制可以把jar和本地库放置在map/reduce任务的classpath或者本地库路径下。

　　分布式缓存设计之初是为了分发一些尺寸不是很大，从几M到几十M的附件。目前实现的分布式缓存的弱点在于不能够指定具体的附件只能应用于特定的map或者reduce。

　　在极少数情况下，由具体任务本身来拷贝其所需的附件要比使用分布式缓存更合适。例如，那种reduce数很少的应用，而且需要的附属文件尺寸非常大（超过512M）。

　　网格模式：应用应该保证分布式缓存中的附件不能够比任务本身的I/O消耗更多。

　　压缩

　　Hadoop Map-Reduce 可以在应用中对map输出的中间数据和reduce的输出数据进行指定的压缩。

• 压缩中间数据： 正如在 shuffle 部分所讲的，对map输出的中间数据进行合适的压缩可以减少map到reduce之间的网络数据传输量，从而提高性能。Lzo 压缩格式是一个压缩map中间数据的合理选择，它有效利用了CPU。
• 压缩应用输出：就如在 输出 部分所讲的， 使用合适的压缩格式压缩输出数据能够减少应用的运行时间。Zlib/Gzip 格式在大多数情况下都是比较适当的选择，因为它在较高压缩率的情况下压缩速度也还算可以，bzip2 就慢得多了。

　　全排序输出

　　抽样

　　有时候，应用需要产生全排序的输出。在这种情况下，一个通用的反模式是只使用一个reduce，这样就能强制数据集中在一处做聚合。很明显，这样做效率不高，这样不仅加大了执行reduce的那个节点的负载，还对失败恢复有严重的不良影响。

　　更好的办法是对输入抽样，然后以此来使用sampling partitioner 代替默认的hash partitioner。这样可以获得更好的负载平衡和失败恢复。#p#

连接（join）有序数据集

　　另一种网格设计模式是关于两个有序数据集的连接，其中一个数据集的大小并非另一个的严格倍数。例如，一个数据集有512个buckets，另一个有200个。

　　在这种情况下，确保输入的数据集是整体有序的（全排序，如同在商议部分所提到的）意味着可以使用两个数据集中的任意一个来作为基数。Pig 就是用这种发发来进行高效的连接。

　　HDFS 操作 & JobTracker 操作

　　NameNode 很重要而且负担要比一般的节点重，所以在进行HDFS 操作的时候要注意对性能的影响。特别是，应用程序不要在map/reduce任务中做非I/O操作，也即像遍历目录，递归统计等这样的元数据操作。

　　同样，不要在应用程序中连接JobTracker来获得关于集群统计的数据。

　　网格模式：应用不应该在代码中执行任何文件系统的元数据操作，这种操作应该在作业提交的时候被严格禁止。除此以外，应用程序不应该自己连接JobTracker 。

　　User Logs

　　与用户执行的任务相关的task-logs，也即 map/reduce 任务的标准输出和错误信息储存在执行这个任务的节点的本地磁盘上。

　　因为每个节点都是共享存储的一部分，所以Map-Reduce 框架对储存在节点上的log数量实际上是有限制的。

　　Web界面

　　Hadoop Map-Reduce 框架提供了一个简单的web界面来监控运行中的作业，查看已完成作业的历史，以及其他一些从JobTracker获得的信息

　　要明白这个web界面是给人看的而不是自动程序。

　　通过一些屏幕自动化软件来从web界面获取信息是不可行的。web界面上的某些部分，像查看历史作业，非常消耗JobTracker 的资源，如果使用屏幕自动化软件这么做可能会导致一些性能问题。

　　假如真有这么一个自动统计汇总的需求，最好去咨询 Map-Reduce的开发团队。

　　工作流

　　Oozie 是一个适用于网格应用的非常好的工作流管理和调度系统。Oozie 可以基于时间或者数据可用性来管理和计划工作流。使用Oozie来管理和调度的低延迟要求的，产品级的项目已经越来越多。

　　设计Oozie 的时候考虑的一个重要因素是Hadoop 更适宜于批量处理大量数据。正因如此，用几个中等规模的Map-Reduce组成处理流程，要比用更多的小型的Map-Reduce作业更好。在极端情况下一个流程可能由几百上千个作业组成，这是很明显的反模式。更好的做法是能够将这些Map-Reduce作业重新组装成较少的几个过程，每个过程处理更多的数据，这有助于提高整个流程的性能并降低延迟。

　　网格模式：工作流中一个Map-Reduce作业应该至少处理十几G数据。

　　反模式

　　这部分会总结一些网格应用通用的反模式。这些东西大多数情况下都与大规模、分布式、批量数据处理系统的原则相悖。这是对应用开发人员的提醒，因为网格软件逐渐规范化固定化，特别是即将发布的20.Fred版本，对于具有下面列出的这些反模式的应用更难容忍。

• 不爱使用像Pig这样的高层次抽象接口。
• 处理几千个小文件（小于1个block的大小，一般是128M），一个map只能处理一个小文件。
• 处理大量数据的，但HDFS block比较小，导致产生数万个map。
• map数量非常多，每个map的运行时间却非常短（例如5秒）。
• 简单聚合却不用Combiner。
• 产生的map数量多于6、7万。
• 处理大数据集的时候只用很少的reduce（例如只用1个）。
• 用Pig 脚本处理大数据集的时候没有用PARALLEL关键字。
• 用1个reduce为所有输出进行全局排序。
• 用很多reduce来处理数据，以致每个reduce只能处理1-2G数据。
• 输出文件多且小。
• 用分布式缓存分发过多的文件或过大的文件（几百M）。
• 一个任务有几十上百个计数器。
• 在map/reduce 任务理执行文件系统元数据操作（例如 listStatus）。
• 用屏幕自动化软件来收集web界面上的信息，作业、队列状态，更糟的是查看已完成作业的历史。
• 工作流由成百上千个小作业做成，每个都只处理少量数据。