CS Five

主题

4.7 分布式数据处理 ​

INFO

来源：4.7 Distributed Data Processing

对应：无

分布式系统常用来收集、访问和处理大型数据集。比如本章前面介绍的数据库系统，就能对分散存储在多台机器上的数据集进行操作。可能没有任何一台机器包含响应查询所需的数据，所以必须靠通信来响应请求。

本节考察一个典型的大数据处理场景：数据集过大，单台机器处理不了，只能分散到多台机器上，每台机器处理其中一部分。处理结果往往还要跨机器汇总，把一台机器的计算结果与其他机器的结果合并起来。为了协调这种分布式数据处理，我们接下来讨论一种叫作 MapReduce 的编程框架。

用 MapReduce 构建分布式数据处理应用，汇集了本书中介绍的许多思想。应用程序用纯函数来表示：先用这些函数映射（map）大型数据集，再把映射得到的值序列归约（reduce）成最终结果。

函数式编程里那些熟悉的概念，在 MapReduce 程序中得到了最大程度的发挥。MapReduce 要求用来映射和归约数据的函数必须是纯函数。一般来说，只用纯函数表示的程序，在执行方式上有很大的灵活性。子表达式可以按任意顺序并行求值，都不会影响最终结果。MapReduce 应用会并行求值许多纯函数，并重新安排计算顺序，以便在分布式系统中高效执行。

MapReduce 的主要优势在于，它在分布式数据处理应用的两个部分之间实现了关注点分离：

  1. 处理数据、合并结果的映射函数和归约函数。
  2. 机器之间的通信和协调。

协调机制负责处理分布式计算中出现的许多问题，比如机器故障、网络故障和进度监控。管理这些问题会给 MapReduce 应用带来一些复杂性，但这些复杂性都不会暴露给应用开发者。相反，构建 MapReduce 应用只需要指定上面第（1）项里的映射函数和归约函数；分布式计算的难题都靠抽象藏了起来。

4.7.1 MapReduce ​

MapReduce 框架把一大批无序的输入值流作为输入，值的类型任意。例如，每个输入可能是某个庞大语料库里的一行文本。计算分三步：

  1. 对每个输入应用映射函数，输出零个或多个任意类型的中间键值对。
  2. 所有中间键值对都按键分组，这样键相同的键值对就能一起归约。
  3. 归约函数把给定键 k 对应的值合并起来；它输出零个或多个值，在最终输出中每个值都与 k 相关联。

为了完成这次计算，MapReduce 框架会创建若干任务（可能分布在不同机器上），各自扮演不同的角色。映射任务（map task）把映射函数应用到输入数据的某个子集上，输出中间键值对。归约（reduce）任务按键对键值对排序、分组，再对每个键的值应用归约函数。映射任务和归约任务之间的所有通信都由框架处理，按键把中间键值对分组这件事也由框架处理。

为了在 MapReduce 应用中利用多台机器，多个映射器在映射阶段（map phase）并行运行，多个归约器在归约阶段（reduce phase）并行运行。两个阶段之间是排序阶段（sort phase），它用排序把键值对归到一起，让所有键相同的键值对都相邻。

考虑统计文本语料库中元音数量的问题。我们可以选用合适的映射函数和归约函数，用 MapReduce 框架来解决它。映射函数接收一行文本作为输入，输出键值对，其中键是元音，值是计数。计数为零的不输出：

python
def count_vowels(line):
    """一个统计一行中元音个数的映射函数。"""
    for vowel in 'aeiou':
        count = line.count(vowel)
        if count > 0:
            emit(vowel, count)

归约函数就是 Python 内置的求和函数，它的输入是值的迭代器（给定键的所有值），返回它们的和。

4.7.2 本地实现 ​

要指定一个 MapReduce 应用，我们需要一份 MapReduce 框架的实现，好把映射函数和归约函数插进去。下一节我们会用开源的 Hadoop 实现；本节先用 Unix 操作系统的内置工具做一个最简实现。

Unix 操作系统在用户程序和计算机底层硬件之间竖起了一道抽象屏障。它提供了一种机制，让程序之间能够通信，尤其是让一个程序能消费另一个程序的输出。在关于 Unix 编程的开创性文章里，Kernigham 和 Pike 断言：「系统的力量更多来自程序之间的关系，而不是程序本身。」

把 Python 源文件变成 Unix 程序的方法，是在第一行加一句注释，说明该用 Python 3 解释器执行。Unix 程序的输入是一个可迭代对象，叫作标准输入（standard input），通过 sys.stdin 访问。迭代它会得到一行行值为字符串的文本。Unix 程序的输出叫作标准输出（standard output），通过 sys.stdout 访问。内置的 print 函数会把一行文本写到标准输出。下面这个 Unix 程序把输入的每一行反向输出：

python
#!/usr/bin/env python3

import sys

for line in sys.stdin:
    print(line.strip('\n')[::-1])

如果把这个程序保存到名为 rev.py 的文件里，就能把它当 Unix 程序执行。首先得告诉操作系统，我们创建了一个可执行程序：

shell
$ chmod u+x rev.py

接下来，我们可以把输入传给这个程序。一个程序的输入可以来自另一个程序。用 | 符号（叫作「管道」）就能做到这一点：它把管道前一个程序的输出送进管道后一个程序。nslookup 程序会输出某个 IP 地址的主机名（这里用的是《纽约时报》的地址）：

shell
$ nslookup 170.149.172.130 | ./rev.py
moc.semityn.www

cat 程序会输出文件的内容。因此可以用 rev.py 程序反转 rev.py 文件的内容：

shell
$ cat rev.py | ./rev.py
3nohtyp vne/nib/rsu/!#

sys tropmi

:nidts.sys ni enil rof
)]1-::[)'n\'(pirts.enil(tnirp

有了这些工具，就足以实现一个基本的 MapReduce 框架。这个版本只有一个映射任务和一个归约任务，两者都是用 Python 实现的 Unix 程序。我们用下面这条命令运行整个 MapReduce 应用：

shell
$ cat input | ./mapper.py | sort | ./reducer.py

mapper.py 和 reducer.py 程序必须实现映射函数和归约函数，还要负责一些简单的输入输出。例如，要实现上面说的元音计数应用，我们会编写下面这个 count_vowels_mapper.py 程序：

python
#!/usr/bin/env python3

import sys
from mr import emit

def count_vowels(line):
    """一个统计一行中元音个数的映射函数。"""
    for vowel in 'aeiou':
        count = line.count(vowel)
        if count > 0:
            emit(vowel, count)

for line in sys.stdin:
    count_vowels(line)

此外，还要编写下面这个 sum_reducer.py 程序：

python
#!/usr/bin/env python3

import sys
from mr import values_by_key, emit

for key, value_iterator in values_by_key(sys.stdin):
    emit(key, sum(value_iterator))

mr module 是本书的配套模块，它提供了 emit 函数和 group_values_by_key 函数：前者用来发射键值对，后者把键相同的值归到一起。该模块还包含 MapReduce 的 Hadoop 分布式实现的接口。

最后，假设我们有下面这个名为 haiku.txt 的输入文件：

shell
Google MapReduce
Is a Big Data framework
For batch processing

用 Unix 管道在本地跑一遍，就能算出这首俳句里每个元音的数量：

shell
$ cat haiku.txt | ./count_vowels_mapper.py | sort | ./sum_reducer.py
'a'   6
'e'   5
'i'   2
'o'   5
'u'   1

4.7.3 分布式实现 ​

Hadoop 是 MapReduce 框架开源实现的名称，能在机器集群上运行 MapReduce 应用，把输入数据和计算分散开来，实现高效的并行处理。它的流接口允许任意的 Unix 程序来定义映射函数和归约函数。事实上，我们的 count_vowels_mapper.py 和 sum_reducer.py 可以直接配合一套 Hadoop 安装使用，来计算大型文本语料库中的元音数量。

和我们简陋的本地 MapReduce 实现相比，Hadoop 有几个优势。第一是速度：不同的机器上同时运行不同的任务，并行应用映射函数和归约函数。第二是容错：任务不管什么原因失败，其结果都可以由另一个任务重新算出，以完成整个计算。第三是监控：框架提供了一个用户界面，用来跟踪 MapReduce 应用的进度。

要用提供的 mapreduce.py 模块运行元音计数应用，请先安装 Hadoop，把 HADOOP 的赋值语句改成你本地安装的根目录，再把一批文本文件复制到 Hadoop 分布式文件系统里，然后运行：

shell
$ python3 mr.py run count_vowels_mapper.py sum_reducer.py [input] [output]

其中 [input] 和 [output] 是 Hadoop 文件系统中的目录。

关于 Hadoop 流接口和系统用法的更多信息，请查阅 Hadoop Streaming Documentation。

基于 MIT 许可发布

版权所有 © 2022-2026 CS Five