MapReduce：大型集群上的简化数据处理

在我们的计算环境中，网络带宽是一个相对稀缺的资源。我们靠充分利用输入数据(由GFS管理【10】)存储在组成集群的机器的本地磁盘中这一事实来节省网络带宽。GFS将每个文件分成64MB的块且在不同机器上存储了每个块的多个副本(通常3个)。MapReeuce master考虑每个输入文件的位置信息且试图调度一台含有相应输入数据的机器上的一个map task。如果失败，它将试图调度与该任务的输入的复制品相邻的一个map task(例如，同一网络交换机中包含相同数据的两台机器)。当在一个集群的 workers 重要部分运行大型MapReduce操作时，大多数输入数据都是本地读取的，并不消耗网络带宽。

3.5 Task粒度

我们将map阶段细分为M个片段，reduce阶段细分为R个片段，如前所述。理想情况下，M和R应该远高于worker机器的数量。每个worker执行多个不同tasks改善了负载均衡，且当一个worker故障后加快了恢复速度：它完成的多个map tasks可以分布到所有其他worker机器上重新执行。

由于master必须做O(M+R)此调度决策和在内存中保持O(M*R)个状态，如前所述，在我们的实现中M和R的数量大小是有实际界限的。(然而，内存的使用量很小。O(M*R)个状态中大约包含每个map/reduce task对一字节的数据。)

此外，R通常受到用户限制，因为每个reduce task的输出最终保存在一个单独的输出文件中。在实践中，我们倾向于选择M因而每个独立task大约有16MB到64MB的输入数据(因而之前所述的局部优化达到最搞笑)，且我们让R是我们希望使用的机器数量的一个小的倍数。我们通常以M=200000, R=5000，使用2000台worker机器执行MapReduce。

3.6 备份Tasks

延长MapReduce操作总时间的一个普遍原因是一个掉队者(straggler)，也就是说，在这个计算中有一台机器花了异常长的时间来完成最后几个map或reduce tasks。掉队者会以一大堆的理由出现。比如说，一台拥有坏磁盘的机器可能经历频繁的矫正错误从而使读取性能从30MB/s降低到了1MB/s。集群调度系统可能在这个机器上调度了其他任务，导致它更慢地执行MapReduce代码，由于竞争CPU、内存、本地磁盘或网络带宽等资源。我们经历的一个最近的问题是机器初始化代码中的一个bug导致处理器缓存失效：受影响的机器计算速度放慢了100倍。

我们有一个通用机制来减轻掉队者问题。当一个MapReduce操作接近完成时，master将调度还在进行的任务的备份执行。无论是原始或者备份执行完成，这个任务都被标记为完成。我们调整了这个机制，因而它增加了该计算的计算资源的使用，但不超过几个百分点。我们发现它大大降低了完成大型MapReduce操作的时间。作为一个例子，当没有备份task机制时，在5.3部分描述的排序程序多花了44%的时间完成。

4 改进

虽然由简单编写的map和reduce函数提供的基本功能已足以满足大多数需求，我们发现了一些有用的扩展。这包括：

• 用户指定的分区(partition)函数来决定如何将中间键值对映射到R个reduce碎片;
• 排序保证：我们的实现保证这R个reduce分区中的每个，中间键值对都按键的升序处理;
• 用户指定的结合(combiner)函数的作用是，在同一个map task内，对按照同一个键生成的中间值进行局部结合，以减少必须在网络间传输的中间数据数量;
• 自定义输入输出类型，为了读新的输入格式和生成新的输出格式;
• 在单机上执行简单debug和小规模测试的一种方式。
• 在【8】中有对这几项的详细讨论。

5 性能表现

在此部分，我们利用大型集群上的两个计算来测量MapReduce的性能表现。一个计算通过搜索大约1TB的数据来找到一个特定的模式。另一个计算对大约1TB的数据进行排序。这两个程序代表由MapReduce用户编写的真正程序的一个大的子集-----程序的一个类用来从一个表示(representation)向另一个表示shuffle数据，另一个类从大数据集中提取小部分关注的数据。

5.1 集群配置

所有程序都在一个拥有大约1800台机器的集群上执行。每台机器拥有两个支持超线程的2GHz的Intel Xeon处理器，4GB内存，两个160GB的IDE磁盘，和千兆以太网接入。这些机器被安排在一个二级树形的交换网络中，该网络根部大约有100~200Gbps的聚合带宽。所有机器都在同一个托管设施中，因此任何一对机器间的往返通信时间不超过1毫秒。

Figure 2: Data transfer rate over time

虽然有4GB内存，但是大约1~1.5GB保留给了运行在集群上的其他任务。这些程序在一个周末的下午执行，此时CPUs，磁盘和网络带宽基本都空闲。

5.2 Grep

grep程序扫描了10^10个100字节的记录，搜索一个相对稀有的三字符模式串(该模式串大约出现在92337个记录中)。输入被划分为了大约64MB大小的片(M=15000)，整个输出都放在了一个文件中(R=1)。

图2展示了计算随时间推移的进展。Y轴显示了输入数据的扫描速率。随着安排到MapReduce计算的机器越来越多，速率也在逐步提升，当安排了1764个workers时速度达到峰值30GB/s以上。map任务结束后，速率来时下降且在大约80秒时到达0。整个计算从开始到结束大约花费了150秒。这包括1分钟的启动消耗。这个消耗来自向所有workers机器传播程序、延迟与GFS的交互以开启一组1000个输入文件，和获取局部优化所需的信息。

（编辑：辽源站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!