第六章 MapReduce 模式

本模式重点介绍 MapReduce 数据处理模式的应用。

MapReduce 是一种数据处理方法，它提供了一种简单的编程模型来处理高度并行化的数据集。它以集群的形式实现，多个节点并行处理数据的不同部分。启动 MapReduce 作业的开销较大，但一旦启动，作业可以快速完成（相对于传统方法而言）。

MapReduce 需要编写两个函数：映射器 (mapper) 和归约器 (reducer)。这些函数接收数据作为输入，然后返回转换后的数据作为输出。这些函数会被反复调用，每次调用都会处理一部分数据。映射器（Mapper）的输出会被聚合，然后发送到归约器（Reducer）。这两个阶段会逐步处理大量数据。

MapReduce 专为批量处理数据集而设计。其限制因素是集群的大小。相同的映射和归约函数可以用于处理非常小的数据集，并且无需随着数据集从千字节增长到兆字节、千兆字节甚至拍字节而进行更改。

MapReduce 可以轻松处理的数据示例包括文本文档（例如维基百科中的所有文档）、Web 服务器日志和用户社交图谱（以便发现新的联系人推荐）。

为了高效处理，数据会被分配到集群中的所有节点上。

背景

MapReduce 模式能够有效应对以下挑战：

• 应用程序处理存储在云端的大量关系型数据
• 应用程序处理存储在云端的大量半结构化或不规则数据
• 应用程序的数据分析需求频繁变化或属于临时性需求
• 应用程序需要生成传统数据库报表工具无法高效生成的报表，因为输入数据量过大或结构不兼容

Hadoop 将 MapReduce 实现为批处理系统。它针对灵活高效地处理海量数据进行了优化，而非响应时间。

MapReduce 的输出结果灵活多变，但通常用于数据挖掘、报表生成，或对数据进行处理以便更传统的报表工具使用。

云的重要性

云平台擅长管理海量数据。大数据分析的原则之一是将计算能力带到数据所在之处，因为移动大量数据既昂贵又缓慢。对于已在云端存储的数据，使用云服务进行分析通常是最具成本效益和效率的。

Hadoop 极大地简化了分布式数据处理流程的构建。在云端运行“Hadoop 即服务”更进一步，简化了 Hadoop 的安装和管理。Hadoop 云服务可以直接访问某些云存储服务中的数据，例如 Amazon 上的 S3 和 Windows Azure 上的 Blob 存储。

通过 Hadoop 云平台服务使用 MapReduce，您可以按需租用实例，缩短使用时间。由于 Hadoop 集群可以包含数百个计算节点，因此可以显著节省成本。当数据存储在与 Hadoop 集成良好的云存储服务中时，这种方式尤其方便。

影响

成本优化、可用性、可扩展性

机制

此模式中实现的 map 和 reduce 函数在概念上与计算机科学中的版本类似，但并不完全相同。在 MapReduce 模式中，列表由键/值对组成，而不仅仅是值。值也可以是多种多样的：文本块、数字，甚至是视频文件。

Hadoop 是一个复杂的框架，可以将 map 和 reduce 函数应用于任意大的数据集。数据被分割成小单元，分布在数据节点集群中。一个典型的 Hadoop 集群包含几个到数百个数据节点。每个数据节点接收数据子集，并根据作业跟踪器的指示，将 map 和 reduce 函数应用于本地存储的数据。作业跟踪器负责协调集群中的作业。在云端，“本地存储”实际上可能指的是持久化的云存储，而非计算节点的本地磁盘驱动器，但原理是相同的。

数据可以在工作流中进行处理，其中会依次应用多组 MapReduce 函数，每对 MapReduce 函数的输出都会成为下一对的输入。最终的输出通常会存储在计算节点的本地磁盘或云存储中。该输出可能是所需的最终结果，也可能只是数据整形操作，用于为 Excel、传统报表工具或商业智能 (BI) 工具等进一步的分析工具做好准备。

MapReduce 应用案例

MapReduce 在许多数据处理工作负载中表现出色，尤其擅长处理那些被称为过易并行问题（embarrassingly parallel problems）。由于数据元素彼此独立，可以按任意顺序处理，因此可以轻松地将过易并行问题并行化。其应用范围非常广泛，虽然本文篇幅有限，无法详尽阐述，但其应用范围涵盖了从 Web 日志文件处理到地震数据分析等诸多领域。

这种模式通常不用于小型数据集，而是用于处理业内所谓的“大数据”。大数据的定义标准尚未完全明确，但通常从数百兆字节或千兆字节开始，一直到PB级。由于MapReduce是一个分布式计算框架，能够简化数据处理，因此可以合理地推断，当数据量过大，单台机器或传统工具无法处理时，大数据便由此开始。

如果待处理的数据量增长到如此规模，则可以先在较小的数据集上开发这种模式，并使其能够持续扩展。从编程角度来看，分析几个小文件和分析分散在数百万个文件中的PB级数据并无本质区别。map和reduce函数无需更改。

超越自定义 MapReduce 函数

Hadoop 支持使用 Java 编写 MapReduce 函数。然而，任何支持标准输入输出 (SIVO) 的编程语言（例如 C++、C#、Python）都可以使用 Hadoop Streams 来实现 MapReduce 函数。此外，根据云环境的具体情况，某些情况下可能还支持其他更高级的编程语言。例如，在 Azure 上的 Hadoop 中，可以使用 JavaScript 编写 Pig 脚本（稍后会详细介绍 Pig）。

Hadoop 不仅仅是一个强大的分布式 MapReduce 引擎。事实上，Apache Hadoop 项目中包含如此之多的库，以至于将 Hadoop 视为一个生态系统更为准确。这个生态系统包含了 MapReduce 之外的更高层次的抽象。

例如，Hive 项目提供了一种类似于传统 SQL 的查询语言抽象；当发出查询时，Hive 会生成 MapReduce 函数并在后台运行它们来执行请求的查询。将 Hive 交互式地用作即席 (ad hoc)查询工具，其体验与使用传统关系数据库类似。然而，由于 Hadoop 是一个批处理环境，其运行速度可能不如传统关系数据库快。

Pig 是另一种查询抽象，它使用一种名为 Pig Latin 的数据流语言。Pig 也会生成 MapReduce 函数，并在后台运行这些函数来实现 Pig Latin 中描述的高级操作。

Mahout 是一种机器学习抽象，负责一些更复杂的任务，例如音乐分类和推荐。与 Hive 和 Pig 一样，Mahout 也会生成 MapReduce 函数并在后台运行它们。

Hive、Pig 和 Mahout 都是抽象层，它们类似于编译器，可以将更高级别的抽象（例如 Java 代码）转换为机器指令。

该生态系统还包含许多其他工具，但并非所有工具都能生成和执行 MapReduce 函数。例如，Sqoop 是一种关系数据库连接器，可让您访问高级的传统数据分析工具。这通常是最有效的组合：使用 Hadoop 获取正确的数据子集并将其塑造成所需的形式，然后使用商业智能工具完成处理。

超越 MapReduce

Hadoop 的功能远不止运行 MapReduce。它是一个高性能操作系统，能够经济高效地构建分布式系统。

为了安全起见，每个字节的数据都会以三份副本的形式存储。这类似于云存储服务通常以三份副本存储数据，但这里指的是 Hadoop 将数据写入其数据节点的本地磁盘驱动器。虽然可以使用云存储来替代本地存储，但这并非必需。

Hadoop 还支持自动故障恢复。如果集群中的某个节点发生故障，它会自动被替换，所有正在运行的作业都会重新启动，并且不会丢失任何数据。此外，Hadoop 还内置了跟踪和监控管理功能。

示例：在 Windows Azure 上构建 PoP

我们希望为 Page of Photos (PoP) 应用程序（前言中已介绍）添加一项新功能：突出显示有史以来访问量最高的页面。为此，我们首先需要页面浏览量数据。这些数据通常记录在 Web 服务器日志中，因此可以轻松解析出来。如水平扩展计算模式（第 2 章）中所述，PoP 的 IIS Web 日志已收集并方便地存储在 Blob 存储中。

我们可以在 Azure 上设置 Hadoop，直接使用 Blob 存储中的 Web 日志文件作为输入。我们需要提供 map 和 reduce 函数来处理 Web 日志文件。这些 map 和 reduce 函数将逐行解析 Web 日志，并从中提取访问过的页面。日志文件的每一行都包含对页面的引用；例如，Web 日志中指示访问 http://www.pageofphotos.com/jebaka 的行将包含字符串“/jebaka”。我们的 map 函数可以移除开头的斜杠字符。我们还可以设置 map 函数忽略所有未访问页面的记录，例如图片下载记录。由于 MapReduce 要求 map 函数返回属性值对，因此我们简单的 map 函数会输出一个字符串，例如“jebaka, 1”，其中“1”表示计数为 1。

MapReduce 会收集所有“jebaka, 1”的实例，并将它们作为一个列表传递给 reduce 函数。这里的键是“jebaka”，传递给 reduce 函数的列表是一个键，后面跟着多个值。reduce 函数的输入将是“jebaka, 1 1 1 1 1 1 1 1 1 1”（依此类推，具体取决于该页面的浏览量）。reduce 函数需要将所有命中次数相加（本例中为 10），并输出“jebaka, 10”，就完成了。

MapReduce 将处理其余的记账工作。最终，会生成一系列包含总计数据的输出文件。虽然可以编写更多 MapReduce 函数来进一步简化流程，但我们假设只需进行简单的文本扫描（不使用 MapReduce）即可找到浏览量最高的页面，并将该值缓存起来，供 PoP 逻辑使用，以便在首页显示最热门的网站。

如果我们想每周更新一次该值，可以安排启动一个 MapReduce 作业。Azure 上的 Hadoop 集群中的各个节点（底层是工作节点实例）每周只会分配很短的时间。

集群能否仅定期启动而不丢失数据，取决于其将数据持久化到云存储的能力。在我们的场景中，Azure 上的 Hadoop 从 Blob 存储读取 Web 日志，并将分析结果写回 Blob 存储。这很方便，并且可以降低计算节点的成本，因为可以按需分配节点来运行 Hadoop 作业，并在作业完成后释放它们。如果数据保存在计算节点的本地磁盘上（传统的非云端 Hadoop 通常就是这样工作的），那么计算节点就需要持续运行。

总结

MapReduce 模式提供了一些简单易用的工具，可以高效地处理任意数量的数据。许多常见应用场景，如果采用传统方法处理，在经济上是不可行的。Hadoop 生态系统提供了更高级的库，简化了复杂 MapReduce 函数的创建和执行。Hadoop 还使得 MapReduce 的输出能够轻松地与其他工具（例如 Excel 和 BI 工具）集成。