Hadoop平台优化综述（二）

4.     从系统实现角度进行优化

4.1    在可移植性和性能之间进行权衡

论文[16]主要针对HDFS进行了优化，它分析了HDFS性能低下的两个原因：调度延迟和可移植性假设。

（1） 调度延迟

Hadoop采用的是动态调度算法，即：当某个tasktracker上出现空slot时，它会通过HEARBEAT（默认时间间隔为3s，当集群变大时，会适当调大）告诉jobtracker，之后jobtracker采用某种调度策略从待选task中选择一个，再通过HEARBEAT告诉tasktracker。从整个过程看，HDFS在获取下一个task之前，一直处于等待状态，这造成了资源利用率不高。此外，由于tasktracker获取新task后，其数据读取过程是完全串行化的，即：tasktracker获取task后，依次连接namenode，连接datanode并读取数据，处理数据。在此过程中，当tasktracker连接namenode和datanode时，HDFS仍在处于等待状态。

为了解决调度延迟问题，可以考虑的解决方案有：重叠I/O和CPU阶段（pipelining），task预取（task prefetching），数据预取（data prefetching）等

（2）可移植性假设

为了增加Hadoop的可移植性，它采用java语言编写，这实际上也潜在的造成了HDFS低效。Java尽管可以让Hadoop的可移植性增强，但是它屏蔽了底层文件系统，这使它没法利用一些底层的API对数据存储和读写进行优化。首先，在共享集群环境下，大量并发读写会增加随机寻道，这大大降低读写效率；另外，并发写会增加磁盘碎片，这将增加读取代价（HDFS适合文件顺序读取）。

为了解决该问题，可以考虑的解决方案有：修改tasktracker上的线程模型，现在Hadoop上的采用的模型是one thread per client，即每个client连接由一个线程处理（包括接受请求，处理请求，返回结果）；修改之后，可将线程分成两组，一组用于处理client通信（Client Thread），一组用于存取数据（Disk Threads，可采用one thread per disk）。

4.2    Prefetching与preshuffling

论文[7]提出了两种优化策略，分别为Prefetching和preshuffling。

(1) PreFetching

preFetching包括Block-intra prefetching和Block-inter prefetching：

Block-intra Prefetching对block内部数据处理方式进行优化。采用的策略是以双向处理（bi-directional processing）方式提升效率，即一端进行计算，一端预取将要用到的数据（同步机制）。

需解决两个问题，一是计算和预取同步。借用进度条（processing bar）的概念，进度条监控两端的进度，当同步将被打破时，调用一个信号。二是确定合适的预取率。通过实验发现，预取数据量并不是越多越好。采用重复实验的方法确定预取数据率。

Block-inter Prefetching在block层面预取数据。当某个task正在处理数据块A1时，预测器预测它接下来要处理的数据块，假设是A2，A3，A4,则将这几个数据块读到task所在的rack上，这样加快了task接下来数据读取速度。

(2) PreShuffling

数据被map task处理之前，由预测器判断每条记录将要被哪个reduce task处理，将这些数据交由靠近该reduce task的节点上的map task处理。

主页： http://incubator.apache.org/projects/hama.html

4.3    Five Factors

论文[8]分析了5个影响Hadoop性能的因素，分别为计算模型，I/O模型，数据解析，索引和调度，同时针对这5个因素提高了相应的提高性能的方法，最后实验证明，通过这些方法可以将Hadoop性能提高2.5到3.5倍。

（1） 计算模型

在Hadoop中，map task产生的中间结果经过sort-merge策略处理后交给reduce task。而这种处理策略（指sort-merge）不能够定制，这对于有些应用而言（有些应用程序可能不需要排序处理），性能不佳。此外，即使是需要排序归并处理的，sort-merge也并不是最好的策略。

本文实现了Fingerprinting Based Grouping（基于hash）策略，该方法明显提高了Hadoop性能。

（2） I/O模型

Reader可以采用两种方式从底层的存储系统中读取数据：direct I/O和streaming I/O。direct I/O是指reader直接从本地文件中读取数据；streaming I/O指使用某种进程间通信方式（如TCP或者JDBC）从另外一个进程中获取数据。从性能角度考虑，direct I/O性能更高，各种数据库系统都是采用direct I/O模式。但从存储独立性考虑，streaming I/O使Hadoop能够从任何进程获取数据，如datanode或database，此外，如果reader不得不从远程节点上读取数据，streaming I/O是仅有的选择。

本文对hadoop的文件读写方式进行了改进，当文件位于本地时，采用direct I/O方式；当文件位于其它节点上时，采用streaming I/O方式。（改进之前，hadoop全是采用streaming I/O方式）。改进后，效率约提高10%。

（3） 数据解析

在hadoop中，原始数据要被转换成key/value的形式以便进一步处理，这就是数据解析。现在有两种数据解析方法：immutable decoding and mutable decoding。Hadoop是采用java语言编写的，java中很多对象是immutable，如String。当用户试图修改一个String内容时，原始对象会被丢弃而新对象会被创建以存储新内容。在Hadoop中，采用了immutable对象存储字符串，这样每解析一个record就会创建一个新的对象，这就导致了性能低下。

本文比较了immutable实现和mutable实现，immutable性能远高于mutable（join是10倍，select是2倍）。

（4） 索引

HDFS设计初衷是处理无结构化数据，既然这样，怎么可能为数据添加索引。实际上，考虑到以下几个因素，仍可以给数据添加索引：

A、 hadoop提供了结构将数据记录解析成key/value对，这样也许可以给key添加索引。

B、 如果作业的输入是一系列索引文件，可以实现一个新的reader高效处理这些文件。

本文设计了一个range 索引，与原系统比较，连接操作提高了大约10倍，选择操作大约提高了2.5倍。

（5） 调度

Hadoop采用的是动态调度策略，即每次调度一个task运行，这样会带来部分开销。而database采用的静态调度的策略，即在编译的时候就确定了调度方案。当用户提交一个sql时，优化器会生成一个分布式查询计划交给每一个节点进行处理。

本文使用一个benchmark评估运行时调度的代价，最终发现运行时调度策略从两个角度影响性能：需要调度的task数；调度算法。对于第一个因素，可以调整block的大小减少task数，对于第二个因素，需要做更多研究，设计新的算法。

本文调整block大小(从64增大到5G)，发现block越大，效率越高，提升性能约20%~30%。

主页： http://www.comp.nus.edu.sg/~epic/

总结

这只是一篇研究性的论文，它只是用实验验证了这5个因素会影响hadoop性能，具体实现不具有通用性，如果想将这5个方面在hadoop中实现，并能够实际的使用，也会还有比较长的距离。

4.4    Hadoop++

论文[9]提出了Hadoop++系统，它为处理结构化或者半结构化数据而设计的，它在Hadoop基础上做了两点改进，一是为HDFS设计了一种索引―Trojan Index。思路是：当数据被加载到HDFS时，自动为每个split建立索引，这样虽然会增加数据加载时的代价，但不影响数据处理过程；二是设计了一种新的join算法―Trojan join。该join算法在数据加载时，将需要join的数据表按照join属性的hash值存放到相同split中，这样只要在map阶段进行局部join便可以得到最终结果，该算法跳过了mapreduce的shuffle和reduce阶段，避免了数据传输的带来的通信代价，因而大大提高了效率。

Hadoop++系统最大的优点是没有直接修改hadoop代码，只是在Hadoop之上提供了供应用程序访问的API。

官方主页： http://infosys.cs.uni-saarland.de/hadoop++.php

5.     Hadoop其它问题

5.1    单点故障问题

Hadoop采用的是C/S架构，因而存在明显的namenode/jobtracker单点故障问题。相比于jobtracker，namenode的单点故障问题更为急迫，因为namenode的故障恢复时间很长，其时间主要花在fsimage加载和blockReport上，下面是一组测试数据：

当前主要的解决思路有：

（1）    Zookeeper。利用分布式系统的可靠协调系统zookeeper维护主从namenode之间的一致性。

（2）    热备。添加热备从namenode，主从namenode之间通过分布式协议维护数据一致性。

（3）    分布式namespace。多个namenode共同管理底层的datanode。

5.2    小文件问题

小文件是指文件size小于HDFS上block大小的文件。这样的文件会给hadoop的扩展性和性能带来严重问题。首先，在HDFS中，任何block，文件或者目录在内存中均以对象的形式存储，每个对象约占150byte，如果有1000 0000个小文件，每个文件占用一个block，则namenode需要2G空间（存两份）。如果存储1亿个文件，则namenode需要20G空间。这样namenode内存容量严重制约了集群的扩展。 其次，访问大量小文件速度远远小于访问几个大文件。HDFS最初是为流式访问大文件开发的，如果访问大量小文件，需要不断的从一个datanode跳到另一个datanode，严重影响性能。最后，处理大量小文件速度远远小于处理同等大小的大文件的速度。每一个小文件要占用一个slot，而task启动将耗费大量时间甚至大部分时间都耗费在启动task和释放task上。

对于Hadoop小文件问题，当前主要有两种解决方案，（1）设计一种工具（比如mapreduce作业）交给用户，让用户自己每隔一段时间将小文件打包成大文件，当前Hadoop本身提供了几个这样的工具，包括Hadoop Archive（Hadoop提供了shell命令），Sequence file（需自己写程序实现）和CombineFileInputFormat（需自己写程序实现）。（2）从系统层面解决HDFS小文件，论文[10][11]介绍了它们思路，大体上说思路基本一致：在原有HDFS基础上添加一个小文件处理模块，当用户上传一个文件时，判断该文件是否属于小文件，如果是，则交给小文件处理模块处理，否则，交给通用文件处理模块处理。小文件处理模块的设计思想是，先将很多小文件合并成一个大文件，然后为这些小文件建立索引，以便进行快速存取和访问。

6.     总结

本文档介绍Hadoop现有的优化点，总体来说，对于Hadoop平台，现在主要有三种优化思路，分别为：从应用程序角度角度进行优化；从参数配置角度进行优化；从系统实现角度进行优化。对于第一种思路，需要根据具体应用需求而定，同时也需要在长期实践中积累和总结；对于第二种思路，大部分采用的方法是根据自己集群硬件和具体应用调整参数，找到一个最优的。对于第三种思路，难度较大，但效果往往非常明显，总结这方面的优化思路，主要有以下几个：

（1）    对 namenode 进行优化，包括增加其吞吐率和解决其单点故障问题。当前主要解决方案有3种：分布式namenode，namenode热备和zookeeper。

（2）    HDFS 小文件问题。当Hadoop中存储大量小文件时，namenode扩展性和性能受到极大制约。现在Hadoop中已有的解决方案包括：Hadoop Archive，Sequence file和CombineFileInputFormat。

（3）    调度框架优化。在Hadoop中，每当出现一个空闲slot后，tasktracker都需要通过HEARBEAT向jobtracker所要task，这个过程的延迟比较大。可以用task预调度的策略解决该问题。

（4）    共享环境下的文件并发存取。在共享环境下，HDFS的随机寻道次数增加，这大大降低了文件存取效率。可以通过优化磁盘调度策略的方法改进。

（5）    索引。索引可以大大提高数据读取效率，如果能根据实际应用需求，为HDFS上的数据添加索引，将大大提高效率。

7.     参考资料

1、 http://developer.yahoo.com/blogs/hadoop/posts/2011/02/mapreduce-nextgen/

2、 http://www.webguo.com/2011/01/18/handoop_job_tuning.html

3、 Optimizing Hadoop Deployments

4、 Baidu Hadoop Extension： https://issues.apache.org/jira/browse/MAPREDUCE-1270

5、 淘宝数据平台与产品部官方博客： http://www.tbdata.org/archives/1423

6、 Shivnath Babu: Towards automatic optimization of MapReduce programs. SoCC 2010: 137-142

7、 Sangwon Seo et al., HPMR: Prefetching and Pre-shuffling SharedMapReduce Computation Environment. In the Proceedings of 11th IEEEInternational Conference on Cluster Computing, Sep. 2009

8、 D. Jiang, B. C. Ooi, L. Shi, S. Wu: The Performance of MapReduce: An In-depth Study. Int’l Conference onVery Large Data Bases (VLDB), 2010

9、 Jens Dittrich, Jorge-Arnulfo Quiane-Ruiz, Alekh Jindal, Yagiz Kargin, Vinay Setty, and Jörg Schad Hadoop++: Making a Yellow Elephant Run Like a Cheetah (Without It Even Noticing)VLDB 2010/PVLDB, Singapore

10、Xuhui Liu, Jizhong Han, Yunqin Zhong, Chengde Han, Xubin He: Implementing WebGIS on Hadoop: A case study of improving small file I/O performance on HDFS. CLUSTER 2009: 1-8

11、Bo Dong, Jie Qiu, Qinghua Zheng, Xiao Zhong, Jingwei Li, Ying Li. A Novel Approach to Improving the Efficiency of Storing and Accessing Small Files on Hadoop: A Case Study by PowerPoint Files. In Proceedings of IEEE SCC’2010. pp.65~72

12、 https://issues.apache.org/jira/browse/HDFS-1052

13、Feng Wang, Jie Qiu, Jie Yang, Bo Dong, Xin Hui Li, Ying Li. Hadoop high availability through metadata replication. In Proceedings of CloudDB’2009. pp.37~44

14、 Rini T. Kaushik, Milind A. Bhandarkar, Klara Nahrstedt. Evaluation and Analysis of GreenHDFS: A Self-Adaptive, Energy-Conserving Variant of the Hadoop Distributed File System. In Proceedings of CloudCom’2010. pp.274~287

15、 Willis Lang, Jignesh M. Patel. Energy Management for MapReduce Clusters. PVLDB, 2010: 129~139

16、Jeffrey Shafer, Scott Rixner, Alan L. Cox: The Hadoop distributed filesystem: Balancing portability and performance. ISPASS 2010: 122-1

17、博文： 7 Tips for Improving MapReduce Performance

――――――――――――――――――――――――――――――――――-

阅读前一篇： 《Hadoop平台优化综述（一）》