[2010-VLDB] The Performance of MapReduce: An In-depth Study

一般认为，云数据处理系统应该能够提供较高的elasticity, scalability, fault tolerance, 而作者提出在上述三个特点之上，云系统也应该提供efficiency。尽管云系统可以通过部署更多的计算节点来提高性能，然而这种做法代价高昂，偏离了云计算的本意。作者提出，一个好的云数据处理系统应当以最经济的方式提供弹性的数据处理。

作者对MapReduce的设计因素进行了深入的研究，从三大方面探讨了五个可能会影响到MapReduce性能的因素：I/O Mode, indexing, data parsing, grouping schemas, block-level scheduling. 作者在Amazon EC2上部署了100个节点进行了相关的实验。 实验结果表明，通过认真地调整这些参数，MapReduce的总体性能能够提高2.5到3.5倍。

1. 关于编程模型

MapReduce编程模型主要关注使用户指定数据的转换逻辑（通过map和reduce函数），并不指定map产生的中间对会怎样被分组以提供给reduce函数处理。为了简化编程复杂性，MapReduce隐藏了中间键值对的分组算法；但是这一算法并不总是最优的，特别是对于那些不关心中间键的顺序的程序，比如aggregation和equal-join。

对于aggregation，作者提出使用fingerprinting based grouping algorithm会更高效一些。该算法为每一个中间键值对的key创建一个32bit的fingerprinting. a 32-bit integer is generated as the fingerprint of the key for each intermediate key/value pair. When a map task sorts the intermediate pairs, it first compares the fingerprints of keys. If two keys have the same fingerprint, we will further compare the original keys. Similarly, when a reduce task merges the collected intermediate pairs, it first groups the pairs by the fingerprints and then for each group, the key-value pairs are merged via the original keys. 第5.10节的实验表明，使用该算法，aggragation task的运行时间能够缩短20%~25%。

对于equal-join，作者提出了多种join方法，其中partition map-side join算法更高效一些。该算法要求把要连接的两个表A和B按照同一个列属性进行相同的分割，比如均按照c属性，从1到100保存为文件A-1和B-1，从101到200保存为A-2和B-2，等等。连接运算只需要一个MapReduce任务，在map中按照A划分，然后每个map去寻找自己对应的B的那块进行连接。因为是自然连接，所以只有两个文件名相同的A表和B表才会连接在一起。然后再Reduce里就可以进行一些别的计算了。

作者说，MapReduce可能需要扩展其编程模型来允许用户指定自己的分组算法。的确，不同的应用可能有着不同的分组需求，一方面要追求编程简单，另一方面又要追求极致性能，如何实现一种折衷的方式，来达到二者之间的平衡，确实是一个值得思考的问题。

2. 存储独立性

MapReduce是独立于存储系统的，它是一个纯粹的数据处理系统，而没有一个built-in的存储引擎；而一般的并行数据库都会将一个查询处理引擎和存储引擎打包在一起。并行数据库读取数据不需要跨引擎访问，可以直接从共享内存中读取数据；而MapReduce需要一个reader从存储系统中读取、解析数据，因此存储独立性的设计在一定程度上影响了MapReduce的性能。

2.1 I/O模式
reader从底层存储系统中读取数据有两种模式：direct IO和stream IO。直接IO比较快，它通过一个DMA控制器可以将数据直接从磁盘缓存读到内存中，而流式IO比较慢，它需要从另一个进程里面读数据。

并行数据库的查询引擎和存储引擎共享内存，采用直接IO的方式；MapReduce采用的是流式IO，比较慢。

作者在HDFS中实现了一种混合IO模式：direct I/O for local reads and streaming I/O for remote reads. 实验结果：direct会比stream快10%，但是相差不大，说明IO模式并不是影响MapReduce性能的主要因素。

2.2 数据解析

reader需要将原始数据转换为键值对。一般decoding schema有两种：immutable schema和mutable schema。其中immutable schema中，object是只读的，如string，为每一个record创建一个单独的object；而mutable schema则是多个record共享一个object。

使用immutable schema慢，因为需要为每一个记录创建一个object，造成CPU的高负载；使用mutable schema，不论有多少个record，仅有一个object，CPU负载较低。

实验结果：mutable 比immutable 要快10倍，而不同数据格式之间的差别并不大（当然，text最快，binary较慢）。immutable 慢主要在于CPU负载，CPU花费80~90%的时间创建object。因此MapReduce性能差不在于数据解析，而在于编码方式，高效编码的关键在于使用mutable decoding schema。

2.3 索引

有三种方式利用索引：

1. 在data split算法中，利用索引来剪枝（主要是根据文件名处理）。

2. 如果要处理的数据是索引格式的话（比如B树），reader可以根据搜索条件得到感兴趣的记录（这不是说提前处理一遍么？会不会增加工作量？）

3. 如果输入是存在n个数据库服务器上的表的话，每个map做一个sql查询。

3. 调度

MapReduce采用运行时调度策略，会带来额外的运行时开销，降低MR job的执行速度；并行数据库系统则采用编译时调度策略，创建执行计划，每个节点知道自己的处理逻辑。MR的运行时调度代价比DBMS的编译时调度要昂贵，然而运行时调度使得MR有灵活的可伸缩性，即在运行时动态地调整资源的能力。

Runtime scheduling 影响性能的因素主要有两方面：

3.1 map的数量

每个InputSplit对应于一个Mapper，每个Data Block对应于一个InputSplit。

实验结果：Block越多，Split越多，Map的数目越少，调度的代价就越小；增加Block size会提高scan性能，但是也会增加错误恢复的代价。512M是平衡点，再大则不会显著提高性能。

3.2 调度算法

Grep实验结果：快的节点结束后会被分配一个位于慢节点上的block，这样就会造成磁盘带宽竞争，从而降低整个数据处理（20%~25%）。

对于MapReduce的调度算法，依然需要进一步的研究。