如何在云环境中实现 Hadoop MapReduce 框架

云计算旨在通过 Internet 提供随需应变的资源或服务，通常视数据中心的规模和可靠性水平而定。MapReduce 是一个为并行处理大量数据而设计的编程模型，它将工作划分为一个独立任务组成的集合。它是一种并行编程，由某种功能随需应变的云（如 Google 的 BigTable、hadoop 和Sector）提供支持。

在本文中，将使用遵从 Randomized Hydrodynamic Load Balancing 技术（下文将详细介绍）的负载平衡算法。利用虚拟化来降低成本和物理服务器的实际数目；更重要的是，将使用虚拟化用来实现高效的物理计算机 CPU 利用。

要从本文获取最多的知识，您应该大致了解云计算的概念、Randomized Hydrodynamic Load Balancing 技术和 Hadoop MapReduce 编程模型。最好对并行程序设计有一个基本的了解，了解 Java™ 或其他面向对象的语言的编程知识将很有帮助。

在本文中，要实现 MapReduce 算法，系统应装有以下软件：

Hadoop 0.20.1.Eclipse IDE 3.0 以上（或 Rational Application Developer 7.1）。Ubuntu 8.2 以上。深入 MapReduce 算法之前，我们将建立基本的云架构、负载平衡、MapReduce 和并行编程 — 至少对本文这是足够了。

云架构：基本内容
图 1 显示了完整的系统详图，包括平台、软件，以及如何使用它们来实现本文的目标设置。

图 1. 云架构

您可以看到，我们使用 Ubuntu 9.04 和 8.2 作为操作系统；平台是 Hadoop 0.20.1、Eclipse 3.3.1 和 Sun Java 6；编程语言使用 Java；脚本语言使用 HTML、JSP 和 XML。

该云架构有一个主节点和一些从属节点。在该实现中，维护主服务器，获取客户端请求并根据请求的类型进行处理。

从图 2 中可以看到，搜索请求转发到 Hadoop NameNode。然后，Hadoop NameNode 负责搜索和索引操作，它将启动大量 Map 和 Reduce 进程。完成特定的搜索关键字 MapReduce 操作之后，NameNode 将输出值返回到服务器并交付客户端。

图 2. Map 和 Reduce 函数执行搜索和索引

如果请求的是特定的软件，则将基于客户租户 ID、支付会费、使用该特定软件的资格以及软件的租赁期完成验证步骤。然后，服务器为该请求提供服务，并允许用户使用特定的软件组合。

在这里提供 SaaS 的多租户功能，单个软件实例可以服务于多个租户。这样，相同的软件映像集合会基于租户 ID 生成不同的实例生成。

这些服务意味着，客户端将使用 Hadoop、Eclipse 等平台以及在搜索文件或使用某些软件时使用的操作系统。此外，为了存储其数据（数据库或文件），在云中，客户端将不得不占用一些数据中心（IaaS）的内存空间。所有这些对终端用户都是透明的。

Randomized Hydrodynamic Load Balancing：基本内容

负载平衡用来确保在使用其他资源时您现有的资源都不是空闲的。若要平衡负载分布，可以将负载从源节点（有多余工作负载）的迁移到相对较轻的负载目标节点。

在运行时应用负载平衡时，它被称之为动态负载平衡— 这可以根据执行节点选择直接实现或以迭代的方式实现：

迭代的方法通过几个迭代步骤确定最终目标节点。直接方法在一个步骤中选择最终目标节点。本文使用 Randomized Hydrodynamic Load Balancing 方法，这是一种同时利用了直接和迭代方法的混合方法。

MapReduce：基本内容

MapReduce 程序用于以并行方式计算大量数据。这就需要在很多计算机之间分配工作负荷。Hadoop 提供了一种系统的方法来实现这种程序设计范式。

计算需要输入一组键/值对，生成一组输出键/值对。计算涉及的两个基本操作：Map 和 Reduce。

用户编写的 Map 操作需要输入并生成一组中间键/值对。MapReduce 库将所有与同一中间键 #1 相关联的中间值组合到一起，并且将它们传递给 Reduce 功能。

同样是用户编写的 Reduce 函数接受一个中间键 #1 和该键的一组值。它将这些值合并形成可能较小的一组值。通常每次 Reduce 调用只生成一个输出值 0 或 1。中间值通过迭代器（一种可以让程序员遍历所有集合元素的对象，忽略其具体的实现）提供给用户的 Reduce 函数。这样，您就可以处理太大而无法放入内存的值列表。

以 WordCount 问题为例。即计算大文件集中的每个单词的匹配次数。Mapper 和 Reducer 函数如代码清单 1 所示。

清单 1. 解决 WordCount 问题的 Map 和 Reduce

[AppleScript] 纯文本查看 复制代码

mapper (filename, file-contents):
  for each word in file-contents:
    emit (word, 1)

reducer (word, values):
  sum = 0
  for each value in values:
    sum = sum + value
  emit (word, sum)

函数发出每个单词附有一个相关的出现次数计数。Reduce 函数对发出的特定单词计数求和。这个基本的功能如果构建在集群之上，就可以很容易转变为高速并行处理系统。

对之前完成的大量数据执行计算，这些数据通常在分布式环境中。Hadoop 的独特之处在于它简单的编程模式 — 使用户可以快速编写和测试分布式系统 — 以及它的高效、自动分配数据和跨计算机处理，进而能够利用 CPU 内核的底层并行度。

让我们理清思路。前文已经讨论过，Hadoop 集群中有以下节点：

NameNode（云主节点）。DataNodes（从属节点）。集群中的节点已经预先加载了本地输入文件。启动 MapReduce 进程时，NameNode 将使用 JobTracker 进程分配任务，它必须由 DataNodes 通过 TaskTracker 进程完成。每个 DataNode 中将运行几个 Map 进程，中间结果会提供给合器进程，生成一台计算机上文件的单词计数（在 WordCount 的问题中）。值被打乱发送到 Reduce 进程，然后生成目标问题的最终输出。

如何使用负载平衡
负载平衡有助于在某个节点的负载超出阈值水平时，将负载平均地分散到空闲的节点。虽然在执行 MapReduce 算法时负载平衡不够明显，但在进行大文件处理以及硬件资源利用至关重要的时候它是非常必要的。一个显著的作用是可以在资源紧张的情况下增加硬件利用率，提高性能。

在某些数据节点已满或者新的空节点加入集群时，实现一个模块来平衡 Hadoop 分布式文件系统集群上的磁盘空间使用情况。达到阈值时将启动平衡器（Class Balancer 工具）；此参数是一个从 0 至 100% 的分数，默认值为 10%。此选项为是否平衡集群设置目标；阈值越小，集群越平衡，同样，平衡器运行的时间也越长。（注：阈值可以很小到您不能平衡集群的状态，因为应用程序可能同时写入和删除文件)

如果对于每个数据节点，节点占用空间与总容量的比率（称为节点利用率）不同于集群已用空间与总空间的比率（集群利用率），且不超过阈值，则认为该集群是平衡的。

该模块将把利用率高的数据节点的数据块以迭代方式移动到利用率低的节点上；在每个迭代中，节点移动或接收不超过容量的阈值比例，每次迭代运行不超过 20 分钟。

在此实现中，节点被分类为高利用、平均利用 和未充分利用。根据每个节点的利用额，将负载在节点之间转移以平衡集群。模块工作方式如下：

首选，它获取邻近节点详细信息：
当 DataNode 的负载增加到阈值级别时，它将向该 NameNode 发送一个请求。NameNode 获得特定 DataNode 最邻近节点的负载级别信息。NameNode 比较负载，然后将有关最空闲相邻节点的详细信息发送到特定的 DataNode。接下来，DataNodes 开始工作：
每个 DataNode 将自己负载量与其最近节点的负载量之和进行比较。如果 DataNode 的负载级别大于其邻近节点，将随机选择那么负载目标节点（直接相邻的节点及其他节点） 。然后将节点请求发送到目标节点。最后，接收请求：
每个节点将维护一个缓冲区接收负载请求。消息传递接口（MPI）管理此缓冲区。主线程会侦听缓冲队列，并服务其接收的请求。节点进入负载平衡执行阶段。

评估性能
提供了不同的输入文件集，每个文件集的大小不同，并在单节点和双节点集群中执行 MapReduce 任务。测量相应的执行时间，我们可以得出结论，到目前为止，在集群中运行 MapReduce 是处理大量输入文件更有效的方式。

图 3 中的关系图说明了我们在各个节点上运行的性能结果。

图 3. MapReduce 负载平衡在集群中更加有效

结束语
我们的 Hadoop MapReduce 和负载平衡的实验可以得到两个必然结论：

在云环境中，MapReduce 结构提高了大型数据集的吞吐效率。相反，您在非云系统中不一定能看到如此的吞吐量增加。 数据集较小时，MapReduce 和负载平衡不会对云系统吞吐量的增加产生明显影响。因此，在计划在云系统上处理大量数据时，考虑 MapReduce 样式