document/ddia

Fork 0

mirror of https://github.com/Vonng/ddia.git synced 2024-12-06 15:20:12 +08:00

Vonng 7f1959e95d ch10 50%

2018-05-13 22:19:48 +08:00

115 KiB

Raw Blame History

10. 批处理

带有太强个人色彩的系统无法成功。当最初的设计完成并且相对稳定时，不同的人们以自己的方式进行测试，真正的考验才开始。

——高德纳

[TOC]

在本书的前两部分中，我们讨论了很多关于请求和查询以及相应的响应或结果。许多现有数据系统中都采用这种数据处理方式：你发送请求指令，一段时间后(我们期望)系统会给出一个结果。数据库，缓存，搜索索引，Web服务器以及其他一些系统都以这种方式工作。

像这样的**在线（online）**系统，无论是浏览器请求页面还是调用远程API的服务，我们通常认为请求是由人类用户触发的，并且正在等待响应。他们不应该等太久，所以我们非常关注系统的响应时间（参阅“描述性能”）。

Web和越来越多的基于HTTP/REST的API使交互的请求/响应风格变得如此普遍，以至于很容易将其视为理所当然。但我们应该记住，这不是构建系统的唯一方式，其他方法也有其优点。我们来看看三种不同类型的系统：

服务（在线系统）

服务等待客户的请求或指令到达。每收到一个，服务会试图尽快处理它，并发回一个响应。响应时间通常是服务性能的主要衡量指标，可用性通常非常重要（如果客户端无法访问服务，用户可能会收到错误消息）。

批处理系统（离线系统）

一个批处理系统有大量的输入数据，跑一个**作业（job）**来处理它，并生成一些输出数据，这往往需要一段时间（从几分钟到几天），所以通常不会有用户等待作业完成。相反，批量作业通常会定期运行（例如，每天一次）。批处理作业的主要性能衡量标准通常是吞吐量（处理特定大小的输入所需的时间）。本章中讨论的就是批处理。

流处理系统（准实时系统）

流处理介于在线和离线（批处理）之间，所以有时候被称为**准实时（near-real-time）或准在线（nearline）**处理。像批处理系统一样，流处理消费输入并产生输出（并不需要响应请求）。但是，流式作业在事件发生后不久就会对事件进行操作，而批处理作业则需等待固定的一组输入数据。这种差异使流处理系统比起批处理系统具有更低的延迟。由于流处理基于批处理，我们将在第11章讨论它。

正如我们将在本章中看到的那样，批处理是构建可靠，可扩展和可维护应用程序的重要组成部分。例如，2004年发布的批处理算法Map-Reduce（可能被过分热情地）被称为“造就Google大规模可扩展性的算法”【2】。随后在各种开源数据系统中得到应用，包括Hadoop，CouchDB和MongoDB。

与多年前为数据仓库开发的并行处理系统【3,4】相比，MapReduce是一个相当低级别的编程模型，但它使得在商用硬件上能进行的处理规模迈上一个新的台阶。虽然MapReduce的重要性正在下降【5】，但它仍然值得去理解，因为它描绘了一幅关于批处理为什么有用，以及如何实用的清晰图景。

实际上，批处理是一种非常古老的计算方式。早在可编程数字计算机诞生之前，打孔卡制表机（例如1890年美国人口普查【6】中使用的霍尔里斯机）实现了半机械化的批处理形式，从大量输入中汇总计算。 Map-Reduce与1940年代和1950年代广泛用于商业数据处理的机电IBM卡片分类机器有着惊人的相似之处【7】。正如我们所说，历史总是在不断重复自己。

在本章中，我们将了解MapReduce和其他一些批处理算法和框架，并探索它们在现代数据系统中的作用。但首先我们将看看使用标准Unix工具的数据处理。即使你已经熟悉了它们，Unix的哲学也值得一读，Unix的思想和经验教训可以迁移到大规模，异构的分布式数据系统中。

使用Unix工具的批处理

我们从一个简单的例子开始。假设您有一台Web服务器，每次处理请求时都会在日志文件中附加一行。例如，使用nginx默认访问日志格式，日志的一行可能如下所示：

216.58.210.78 - - [27/Feb/2015:17:55:11 +0000] "GET /css/typography.css HTTP/1.1" 
200 3377 "http://martin.kleppmann.com/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_5) 
AppleWebKit/537.36 (KHTML, like Gecko) Chrome/40.0.2214.115 Safari/537.36"

（实际上这只是一行，分成多行只是为了便于阅读。）这一行中有很多信息。为了解释它，你需要了解日志格式的定义，如下所示：

 $remote_addr - $remote_user [$time_local] "$request"
 $status $body_bytes_sent "$http_referer" "$http_user_agent"

日志的这一行表明在2015年2月27日17:55:11 UTC，服务器从客户端IP地址216.58.210.78接收到对文件/css/typography.css的请求。用户没有被认证，所以$remote_user被设置为连字符（- ）。响应状态是200（即请求成功），响应的大小是3377字节。网页浏览器是Chrome 40，URL http://martin.kleppmann.com/ 的页面中的引用导致该文件被加载。

分析简单日志

很多工具可以从这些日志文件生成关于网站流量的漂亮的报告，但为了练手，让我们使用基本的Unix功能创建自己的工具。例如，假设你想在你的网站上找到五个最受欢迎的网页。则可以在Unix shell中这样做：¹

cat /var/log/nginx/access.log | #1
	awk '{print $7}' | #2
	sort             | #3
	uniq -c          | #4
	sort -r -n       | #5
	head -n 5          #6

读取日志文件
将每一行按空格分割成不同的字段，每行只输出第七个字段，恰好是请求的URL。在我们的例子中是/css/typography.css。
按字母顺序排列请求的URL列表。如果某个URL被请求过n次，那么排序后，文件将包含连续重复出现n次的该URL。
uniq命令通过检查两个相邻的行是否相同来过滤掉输入中的重复行。 -c则表示还要输出一个计数器：对于每个不同的URL，它会报告输入中出现该URL的次数。
第二种排序按每行起始处的数字（-n）排序，这是URL的请求次数。然后逆序（-r）返回结果，大的数字在前。
最后，只输出前五行（-n 5），并丢弃其余的。该系列命令的输出如下所示：

    4189 /favicon.ico
    3631 /2013/05/24/improving-security-of-ssh-private-keys.html
    2124 /2012/12/05/schema-evolution-in-avro-protocol-buffers-thrift.html
    1369 /
     915 /css/typography.css

如果你不熟悉Unix工具，上面的命令行可能看起来有点吃力，但是它非常强大。它能在几秒钟内处理几GB的日志文件，并且您可以根据需要轻松修改命令。例如，如果要从报告中省略CSS文件，可以将awk参数更改为'$7 !~ /\.css$/ {print $7}',如果想统计最多的客户端IP地址,可以把awk参数改为'{print $1}'等等。

我们不会在这里详细探索Unix工具，但是它非常值得学习。令人惊讶的是，使用awk，sed，grep，sort，uniq和xargs的组合，可以在几分钟内完成许多数据分析，并且它们的性能相当的好【8】。

命令链与自定义程序

除了Unix命令链，你还可以写一个简单的程序来做同样的事情。例如在Ruby中，它可能看起来像这样：

counts = Hash.new(0)         # 1
File.open('/var/log/nginx/access.log') do |file| 
    file.each do |line|
        url = line.split[6]  # 2
        counts[url] += 1     # 3
    end
end

top5 = counts.map{|url, count| [count, url] }.sort.reverse[0...5] # 4
top5.each{|count, url| puts "#{count} #{url}" }                   # 5

counts是一个存储计数器的哈希表，保存了每个URL被浏览的次数，默认为0。
逐行读取日志，抽取每行第七个被空格分隔的字段为URL（这里的数组索引是6，因为Ruby的数组索引从0开始计数）
将日志当前行中URL对应的计数器值加一。
按计数器值（降序）对哈希表内容进行排序，并取前五位。
打印出前五个条目。

这个程序并不像Unix管道那样简洁，但是它的可读性很强，喜欢哪一种属于口味的问题。但两者除了表面上的差异之外，执行流程也有很大差异，如果你在大文件上运行此分析，则会变得明显。

排序 VS 内存中的聚合

Ruby脚本在内存中保存了一个URL的哈希表，将每个URL映射到它出现的次数。 Unix管道没有这样的哈希表，而是依赖于对URL列表的排序，在这个URL列表中，同一个URL的只是简单地重复出现。

哪种方法更好？这取决于你有多少个不同的URL。对于大多数中小型网站，你可能可以为所有不同网址提供一个计数器（假设我们使用1GB内存）。在此例中，作业的工作集（working set）（作业需要随机访问的内存大小）仅取决于不同URL的数量：如果日志中只有单个URL，重复出现一百万次，则散列表所需的空间表就只有一个URL加上一个计数器的大小。当工作集足够小时，内存散列表表现良好，甚至在性能较差的笔记本电脑上也可以正常工作。

另一方面，如果作业的工作集大于可用内存，则排序方法的优点是可以高效地使用磁盘。这与我们在“SSTables和LSM树”中讨论过的原理是一样的：数据块可以在内存中排序并作为段文件写入磁盘，然后多个排序好的段可以合并为一个更大的排序文件。归并排序具有在磁盘上运行良好的顺序访问模式。（请记住，针对顺序I/O进行优化是第3章中反复出现的主题，相同的模式在此重现）

GNU Coreutils（Linux）中的sort 程序通过溢出至磁盘的方式来自动应对大于内存的数据集，并能同时使用多个CPU核进行并行排序【9】。这意味着我们之前看到的简单的Unix命令链很容易扩展到大数据集，且不会耗尽内存。瓶颈可能是从磁盘读取输入文件的速度。

Unix哲学

我们可以非常容易地使用前一个例子中的一系列命令来分析日志文件，这并非巧合：事实上，这实际上是Unix的关键设计思想之一，且它今天仍然令人讶异地关联。让我们更深入地研究一下，以便从Unix中借鉴一些想法【10】。

Unix管道的发明者道格·麦克罗伊（Doug McIlroy）在1964年首先描述了这种情况【11】：“当我们需要将消息从一个程序传递另一个程序时，我们需要一种类似水管法兰的拼接程序的方式【a】，I/O应该也按照这种方式进行“。水管的类比仍然在生效，通过管道连接程序的想法成为了现在被称为Unix哲学的一部分 —— 这一组设计原则在Unix用户与开发者之间流行起来，该哲学在1978年表述如下【12,13】：

让每个程序都做好一件事。要做一件新的工作，写一个新程序，而不是通过添加“功能”让老程序复杂化。
期待每个程序的输出成为另一个程序的输入。不要将无关信息混入输出。避免使用严格的列数据或二进制输入格式。不要坚持交互式输入。
设计和构建软件，甚至是操作系统，要尽早尝试，最好在几周内完成。不要犹豫，扔掉笨拙的部分，重建它们。
优先使用工具来减轻编程任务，即使必须曲线救国编写工具，且在用完后很可能要扔掉大部分。

这种方法 —— 自动化，快速原型设计，增量式迭代，对实验友好，将大型项目分解成可管理的块 —— 听起来非常像今天的敏捷开发和DevOps运动。奇怪的是，四十年来变化不大。

sort工具是一个很好的例子。可以说它比大多数编程语言标准库中的实现（即使有很大好处，也不会溢出到磁盘或使用多线程）要更好。然而，单独使用sort 几乎没什么用。它只能与其他Unix工具（如uniq）结合使用。

像 bash这样的Unix shell可以让我们轻松地将这些小程序组合成令人讶异的强大数据处理任务。尽管这些程序中有很多是由不同人群编写的，但它们可以灵活地结合在一起。 Unix如何实现这种可组合性？

统一的接口

如果你希望一个程序的输出成为另一个程序的输入，那意味着这些程序必须使用相同的数据格式 —— 换句话说，一个兼容的接口。如果你希望能够将任何程序的输出连接到任何程序的输入，那意味着所有程序必须使用相同的I/O接口。

在Unix中，这种接口是一个文件（file）（更准确地说，是一个文件描述符）。一个文件只是一串有序的字节序列。因为这是一个非常简单的接口，所以可以使用相同的接口来表示许多不同的东西：文件系统上的真实文件，到另一个进程（Unix套接字，stdin，stdout）的通信通道，设备驱动程序（比如/dev/audio或/dev/lp0），表示TCP连接的套接字等等。很容易将这些设计视为理所当然的，但实际上能让这些差异巨大的东西共享一个统一的接口是非常厉害的，这使得它们可以很容易地连接在一起²。

按照惯例，许多（但不是全部）Unix程序将这个字节序列视为ASCII文本。我们的日志分析示例使用了这个事实：awk，sort，uniq和head都将它们的输入文件视为由\n（换行符，ASCII 0x0A）字符分隔的记录列表。 \n的选择是任意的 —— 可以说，ASCII记录分隔符0x1E本来就是一个更好的选择，因为它是为了这个目的而设计的【14】，但是无论如何，所有这些程序都使用相同的记录分隔符允许它们互操作。

每条记录（即一行输入）的解析则更加模糊。 Unix工具通常通过空白或制表符将行分割成字段，但也使用CSV（逗号分隔），管道分隔和其他编码。即使像xargs这样一个相当简单的工具也有六个命令行选项，用于指定如何解析输入。

ASCII文本的统一接口大多数时候都能工作，但它不是很优雅：我们的日志分析示例使用{print $7}来提取网址，这样可读性不是很好。在理想的世界中可能是{print $request_url}或类似的东西。我们稍后会回顾这个想法。

尽管几十年后还不够完美，但统一的Unix接口仍然是非常出色的设计。没有多少软件能像Unix工具一样交互组合的这么好：你不能通过自定义分析工具轻松地将电子邮件帐户的内容和在线购物历史记录以管道传送至电子表格中，并将结果发布到社交网络或维基。今天，像Unix工具一样流畅地运行程序是一种例外，而不是规范。

即使是具有相同数据模型的数据库，将数据从一种导出再导入另一种也并不容易。缺乏整合导致了数据的巴尔干化³。

逻辑与布线相分离

Unix工具的另一个特点是使用标准输入（stdin）和标准输出（stdout）。如果你运行一个程序，而不指定任何其他的东西，标准输入来自键盘，标准输出指向屏幕。但是，你也可以从文件输入和/或将输出重定向到文件。管道允许你将一个进程的标准输出附加到另一个进程的标准输入（有个小内存缓冲区，而不需要将整个中间数据流写入磁盘）。

程序仍然可以直接读取和写入文件，但如果程序不担心特定的文件路径，只使用标准输入和标准输出，则Unix方法效果最好。这允许shell用户以任何他们想要的方式连接输入和输出；该程序不知道或不关心输入来自哪里以及输出到哪里。（人们可以说这是一种松耦合（loose coupling），晚期绑定（late binding）【15】或控制反转（inversion of control）【16】）。将输入/输出布线与程序逻辑分开，可以将小工具组合成更大的系统。

你甚至可以编写自己的程序，并将它们与操作系统提供的工具组合在一起。你的程序只需要从标准输入读取输入，并将输出写入标准输出，它就可以加入数据处理的管道中。在日志分析示例中，你可以编写一个将Usage-Agent字符串转换为更灵敏的浏览器标识符，或者将IP地址转换为国家代码的工具，并将其插入管道。sort程序并不关心它是否与操作系统的另一部分或者你写的程序通信。

但是，使用stdin和stdout能做的事情是有限的。需要多个输入或输出的程序是可能的，但非常棘手。你没法将程序的输出管道连接至网络连接中【17,18】⁴ 。如果程序直接打开文件进行读取和写入，或者将另一个程序作为子进程启动，或者打开网络连接，那么I/O的布线就取决于程序本身了。它仍然可以被配置（例如通过命令行选项），但在Shell中对输入和输出进行布线的灵活性就少了。

透明度和实验

使Unix工具如此成功的部分原因是，它们使查看正在发生的事情变得非常容易：

Unix命令的输入文件通常被视为不可变的。这意味着你可以随意运行命令，尝试各种命令行选项，而不会损坏输入文件。
你可以在任何时候结束管道，将管道输出到less，然后查看它是否具有预期的形式。这种检查能力对调试非常有用。
你可以将一个流水线阶段的输出写入文件，并将该文件用作下一阶段的输入。这使你可以重新启动后面的阶段，而无需重新运行整个管道。

因此，与关系数据库的查询优化器相比，即使Unix工具非常简单，但仍然非常有用，特别是对于实验而言。

然而，Unix工具的最大局限在于它们只能在一台机器上运行 —— 而Hadoop这样的工具即应运而生。

MapReduce和分布式文件系统

MapReduce有点像Unix工具，但分布在数千台机器上。像Unix工具一样，它相当简单粗暴，但令人惊异地管用。一个MapReduce作业可以和一个Unix进程相类比：它接受一个或多个输入，并产生一个或多个输出。

和大多数Unix工具一样，运行MapReduce作业通常不会修改输入，除了生成输出外没有任何副作用。输出文件以连续的方式一次性写入（一旦写入文件，不会修改任何现有的文件部分）。

虽然Unix工具使用stdin和stdout作为输入和输出，但MapReduce作业在分布式文件系统上读写文件。在Hadoop的Map-Reduce实现中，该文件系统被称为HDFS（Hadoop分布式文件系统），一个Google文件系统（GFS）的开源实现【19】。

除HDFS外，还有各种其他分布式文件系统，如GlusterFS和Quantcast File System（QFS）【20】。诸如Amazon S3，Azure Blob存储和OpenStack Swift 【21】等对象存储服务在很多方面都是相似的⁵。在本章中，我们将主要使用HDFS作为示例，但是这些原则适用于任何分布式文件系统。

与网络连接存储（NAS）和存储区域网络（SAN）架构的共享磁盘方法相比，HDFS基于无共享原则（参见第二部分前言）。共享磁盘存储由集中式存储设备实现，通常使用定制硬件和专用网络基础设施（如光纤通道）。而另一方面，无共享方法不需要特殊的硬件，只需要通过传统数据中心网络连接的计算机。

HDFS包含在每台机器上运行的守护进程，对外暴露网络服务，允许其他节点访问存储在该机器上的文件（假设数据中心中的每台通用计算机都挂载着一些磁盘）。名为NameNode的中央服务器会跟踪哪个文件块存储在哪台机器上。因此，HDFS在概念上创建了一个大型文件系统，可以使用所有运行有守护进程的机器的磁盘。

为了容忍机器和磁盘故障，文件块被复制到多台机器上。复制可能意味着多个机器上的相同数据的多个副本，如第5章中所述，或者诸如Reed-Solomon码这样的纠删码方案，它允许以比完全复制更低的存储开销以恢复丢失的数据【20,22】。这些技术与RAID相似，可以在连接到同一台机器的多个磁盘上提供冗余；区别在于在分布式文件系统中，文件访问和复制是在传统的数据中心网络上完成的，没有特殊的硬件。

HDFS已经扩展的很不错了：在撰写本书时，最大的HDFS部署运行在上万台机器上，总存储容量达数百PB【23】。如此大的规模已经变得可行，因为使用商品硬件和开源软件的HDFS上的数据存储和访问成本远低于专用存储设备上的同等容量【24】。

MapReduce作业执行

MapReduce是一个编程框架，你可以使用它编写代码来处理HDFS等分布式文件系统中的大型数据集。理解它的最简单方法是参考“简单日志分析”中的Web服务器日志分析示例。MapReduce中的数据处理模式与此示例非常相似：

读取一组输入文件，并将其分解成记录（records）。在Web服务器日志示例中，每条记录都是日志中的一行（即\n是记录分隔符）。
调用Mapper函数，从每条输入记录中提取一对键值。在前面的例子中，Mapper函数是awk '{print $7}'：它提取URL（$7）作为关键字，并将值留空。
按键排序所有的键值对。在日志的例子中，这由第一个sort命令完成。
调用Reducer函数遍历排序后的键值对。如果同一个键出现多次，排序使它们在列表中相邻，所以很容易组合这些值而不必在内存中保留很多状态。在前面的例子中，Reducer是由uniq -c命令实现的，该命令使用相同的键来统计相邻记录的数量。

这四个步骤可以作为一个MapReduce作业执行。步骤2（Map）和4（Reduce）是你编写自定义数据处理代码的地方。步骤1（将文件分解成记录）由输入格式解析器处理。步骤3中的排序步骤隐含在MapReduce中 —— 你不必编写它，因为Mapper的输出始终在送往Reducer之前进行排序。

要创建MapReduce作业，你需要实现两个回调函数，Mapper和Reducer，其行为如下（参阅“MapReduce查询”）：

Mapper

Mapper会在每条输入记录上调用一次，其工作是从输入记录中提取键值。对于每个输入，它可以生成任意数量的键值对（包括None）。它不会保留从一个输入记录到下一个记录的任何状态，因此每个记录都是独立处理的。

Reducer MapReduce框架拉取由Mapper生成的键值对，收集属于同一个键的所有值，并使用在这组值列表上迭代调用Reducer。 Reducer可以产生输出记录（例如相同URL的出现次数）。

在Web服务器日志的例子中，我们在第5步中有第二个sort命令，它按请求数对URL进行排序。在MapReduce中，如果你需要第二个排序阶段，则可以通过编写第二个MapReduce作业并将第一个作业的输出用作第二个作业的输入来实现它。这样看来，Mapper的作用是将数据放入一个适合排序的表单中，并且Reducer的作用是处理已排序的数据。

分布式执行MapReduce

MapReduce与Unix命令管道的主要区别在于，MapReduce可以在多台机器上并行执行计算，而无需编写代码来显式处理并行问题。Mapper和Reducer一次只能处理一条记录；它们不需要知道它们的输入来自哪里，或者输出去往什么地方，所以框架可以处理在机器之间移动数据的复杂性。

在分布式计算中可以使用标准的Unix工具作为Mapper和Reducer【25】，但更常见的是，它们被实现为传统编程语言的函数。在Hadoop MapReduce中，Mapper和Reducer都是实现特定接口的Java类。在MongoDB和CouchDB中，Mapper和Reducer都是JavaScript函数（参阅“MapReduce查询”）。

图10-1显示了Hadoop MapReduce作业中的数据流。其并行化基于分区（参见第6章）：作业的输入通常是HDFS中的一个目录，输入目录中的每个文件或文件块都被认为是一个单独的分区，可以单独处理map任务（图10-1中的m1，m2和m3标记）。

每个输入文件的大小通常是数百兆字节。 MapReduce调度器（图中未显示）试图在其中一台存储输入文件副本的机器上运行每个Mapper，只要该机器有足够的备用RAM和CPU资源来运行Mapper任务【26】。这个原则被称为将计算放在数据附近【27】：它节省了通过网络复制输入文件的开销，减少网络负载并增加局部性。

图10-1 具有三个Mapper和三个Reducer的MapReduce任务

在大多数情况下，应该在Mapper任务中运行的应用代码在将要运行它的机器上还不存在，所以MapReduce框架首先将代码（例如Java程序中的JAR文件）复制到适当的机器。然后启动Map任务并开始读取输入文件，一次将一条记录传入Mapper回调函数。Mapper的输出由键值对组成。

计算的Reduce端也被分区。虽然Map任务的数量由输入文件块的数量决定，但Reducer的任务的数量是由作业作者配置的（它可以不同于Map任务的数量）。为了确保具有相同键的所有键值对最终落在相同的Reducer处，框架使用键的散列值来确定哪个Reduce任务应该接收到特定的键值对（参见“按键散列分区”））。

键值对必须进行排序，但数据集可能太大，无法在单台机器上使用常规排序算法进行排序。相反，分类是分阶段进行的。首先每个Map任务都按照Reducer对输出进行分区。每个分区都被写入Mapper程序的本地磁盘，使用的技术与我们在“SSTables与LSM树”中讨论的类似。

只要当Mapper读取完输入文件，并写完排序后的输出文件，MapReduce调度器就会通知Reducer可以从该Mapper开始获取输出文件。Reducer连接到每个Mapper，并下载自己相应分区的有序键值对文件。按Reducer分区，排序，从Mapper向Reducer复制分区数据，这一整个过程被称为混洗（shuffle）【26】（一个容易混淆的术语 —— 不像洗牌，在MapReduce中的混洗没有随机性）。

Reduce任务从Mapper获取文件，并将它们合并在一起，并保留有序特性。因此，如果不同的Mapper生成了键相同的记录，则在Reducer的输入中，这些记录将会相邻。

Reducer调用时会收到一个键，和一个迭代器作为参数，迭代器会顺序地扫过所有具有该键的记录（因为在某些情况可能无法完全放入内存中）。Reducer可以使用任意逻辑来处理这些记录，并且可以生成任意数量的输出记录。这些输出记录会写入分布式文件系统上的文件中（通常是在跑Reducer的机器本地磁盘上留一份，并在其他机器上留几份副本）。

MapReduce工作流

单个MapReduce作业可以解决的问题范围很有限。以日志分析为例，单个MapReduce作业可以确定每个URL的页面浏览次数，但无法确定最常见的URL，因为这需要第二轮排序。

因此将MapReduce作业链接成为**工作流（workflow）**中是极为常见的，例如，一个作业的输出成为下一个作业的输入。 Hadoop Map-Reduce框架对工作流没有特殊支持，所以这个链是通过目录名隐式实现的：第一个作业必须将其输出配置为HDFS中的指定目录，第二个作业必须将其输入配置为从同一个目录。从MapReduce框架的角度来看，这是是两个独立的作业。

因此，被链接的MapReduce作业并没有那么像Unix命令管道（它直接将一个进程的输出作为另一个进程的输入，仅用一个很小的内存缓冲区）。它更像是一系列命令，其中每个命令的输出写入临时文件，下一个命令从临时文件中读取。这种设计有利也有弊，我们将在“物化中间状态”中讨论。

只有当作业成功完成后，批处理作业的输出才会被视为有效的（MapReduce会丢弃失败作业的部分输出）。因此，工作流中的一项作业只有在先前的作业 —— 即生产其输入的作业 —— 成功完成后才能开始。为了处理这些作业之间的依赖，有很多针对Hadoop的工作流调度器被开发出来，包括Oozie，Azkaban，Luigi，Airflow和Pinball 【28】。

这些调度程序还具有管理功能，在维护大量批处理作业时非常有用。在构建推荐系统时，由50到100个MapReduce作业组成的工作流是常见的【29】。而在大型组织中，许多不同的团队可能运行不同的作业来读取彼此的输出。工具支持对于管理这样复杂的数据流而言非常重要。

Hadoop的各种高级工具（如Pig 【30】，Hive 【31】，Cascading 【32】，Crunch 【33】和FlumeJava 【34】）也能自动布线组装多个MapReduce阶段，生成合适的工作流。

Reduce端连接与分组

我们在第2章中讨论了数据模型和查询语言的联接，但是我们还没有深入探讨连接是如何实现的。现在是我们再次捡起这条线索的时候了。

在许多数据集中，一条记录与另一条记录存在关联是很常见的：关系模型中的外键，文档模型中的文档引用或图模型中的边。当你需要同时访问这一关联的两侧（持有引用的记录与被引用的记录）时，连接就是必须的。（包含引用的记录和被引用的记录），连接就是必需的。正如第2章所讨论的，非规范化可以减少对连接的需求，但通常无法将其完全移除⁶。

在数据库中，如果执行只涉及少量记录的查询，数据库通常会使用索引来快速定位感兴趣的记录（参阅第3章）。如果查询涉及到连接，则可能涉及到查找多个索引。然而MapReduce没有索引的概念 —— 至少在通常意义上没有。

当MapReduce作业被赋予一组文件作为输入时，它读取所有这些文件的全部内容；数据库会将这种操作称为全表扫描。如果你只想读取少量的记录，则全表扫描与索引查询相比，代价非常高昂。但是在分析查询中（参阅“事务处理或分析？”），通常需要计算大量记录的聚合。在这种情况下，特别是如果能在多台机器上并行处理时，扫描整个输入可能是相当合理的事情。

当我们在批处理的语境中讨论连接时，我们指的是在数据集中解析某种关联的全量存在。例如我们假设一个作业是同时处理所有用户的数据，而非仅仅是为某个特定用户查找数据（而这能通过索引更高效地完成）。

示例：分析用户活动事件

图10-2给出了一个批处理作业中连接的典型例子。左侧是事件日志，描述登录用户在网站上做的事情（称为活动事件（activity events）或点击流数据（clickstream data）），右侧是用户数据库。你可以将此示例看作是星型模式的一部分（参阅“星型和雪花型：分析的模式”）：事件日志是事实表，用户数据库是其中的一个维度。

图10-2 用户行为日志与用户档案的连接

分析任务可能需要将用户活动与用户简档相关联：例如，如果档案包含用户的年龄或出生日期，系统就可以确定哪些页面更受哪些年龄段的用户欢迎。然而活动事件仅包含用户ID，而没有包含完整的用户档案信息。在每个活动事件中嵌入这些档案信息很可能会非常浪费。因此，活动事件需要与用户档案数据库相连接。

实现这一连接的最简单方法是，逐个遍历活动事件，并为每个遇到的用户ID查询用户数据库（在远程服务器上）。这是可能的，但是它的性能可能会非常差：处理吞吐量将受限于受数据库服务器的往返时间，本地缓存的有效性很大程度上取决于数据的分布，并行运行大量查询可能会轻易压垮数据库【35】。

为了在批处理过程中实现良好的吞吐量，计算必须（尽可能）限于单台机器上进行。为待处理的每条记录发起随机访问的网络请求实在是太慢了。而且，查询远程数据库意味着批处理作业变为非确定的（nondeterministic），因为远程数据库中的数据可能会改变。

因此，更好的方法是获取用户数据库的副本（例如，使用ETL进程从数据库备份中提取数据，参阅“数据仓库”），并将它和用户行为日志放入同一个分布式文件系统中。然后你可以将用户数据库存储在HDFS中的一组文件中，而用户活动记录存储在另一组文件中，并能用MapReduce将所有相关记录集中到同一个地方进行高效处理。

排序合并连接

回想一下，Mapper的目的是从每个输入记录中提取一对键值。在图10-2的情况下，这个键就是用户ID：一组Mapper会扫过活动事件（提取用户ID作为键，活动事件作为值），而另一组Mapper将会扫过用户数据库（提取用户ID作为键，用户的出生日期作为值）。这个过程如图10-3所示。

图10-3 在用户ID上进行的Reduce端连接。如果输入数据集分区为多个文件，则每个分区都会被多个Mapper并行处理

当MapReduce框架通过键对Mapper输出进行分区，然后对键值对进行排序时，效果是具有相同ID的所有活动事件和用户记录在Reducer输入中彼此相邻。 Map-Reduce作业甚至可以也让这些记录排序，使Reducer总能先看到来自用户数据库的记录，紧接着是按时间戳顺序排序的活动事件 —— 这种技术被称为二次排序（secondary sort）【26】。

然后Reducer可以容易地执行实际的连接逻辑：每个用户ID都会被调用一次Reducer函数，且因为二次排序，第一个值应该是来自用户数据库的出生日期记录。 Reducer将出生日期存储在局部变量中，然后使用相同的用户ID遍历活动事件，输出已观看网址和观看者年龄的结果对。随后的Map-Reduce作业可以计算每个URL的查看者年龄分布，并按年龄段进行聚集。

由于Reducer一次处理一个特定用户ID的所有记录，因此一次只需要将一条用户记录保存在内存中，而不需要通过网络发出任何请求。这个算法被称为排序合并连接（sort-merge join），因为Mapper的输出是按键排序的，然后Reducer将来自连接两侧的有序记录列表合并在一起。

把相关数据放在一起

在排序合并连接中，Mapper和排序过程确保了所有对特定用户ID执行连接操作的必须数据都被放在同一个地方：单次调用Reducer的地方。预先排好了所有需要的数据，Reducer可以是相当简单的单线程代码，能够以高吞吐量和与低内存开销扫过这些记录。

这种架构可以看做，Mapper将“消息”发送给Reducer。当一个Mapper发出一个键值对时，这个键的作用就像值应该传递到的目标地址。即使键只是一个任意的字符串（不是像IP地址和端口号那样的实际的网络地址），它表现的就像一个地址：所有具有相同键的键值对将被传递到相同的目标（一次Reduce的调用）。

使用MapReduce编程模型，能将计算的物理网络通信层面（从正确的机器获取数据）从应用逻辑中剥离出来（获取数据后执行处理）。这种分离与数据库的典型用法形成了鲜明对比，从数据库中获取数据的请求经常出现在应用代码内部【36】。由于MapReduce能够处理所有的网络通信，因此它也避免了应用代码去担心部分故障，例如另一个节点的崩溃：MapReduce在不影响应用逻辑的情况下能透明地重试失败的任务。

GROUP BY

除了连接之外，“把相关数据放在一起”的另一种常见模式是，按某个键对记录分组（如SQL中的GROUP BY子句）。所有带有相同键的记录构成一个组，而下一步往往是在每个组内进行某种聚合操作，例如：

统计每个组中记录的数量（例如在统计PV的例子中，在SQL中表示为COUNT(*)聚合）
对某个特定字段求和（SQL中的SUM(fieldname)）
按某种分级函数取出排名前k条记录。

使用MapReduce实现这种分组操作的最简单方法是设置Mapper，以便它们生成的键值对使用所需的分组键。然后分区和排序过程将所有具有相同分区键的记录导向同一个Reducer。因此在MapReduce之上实现分组和连接看上去非常相似。

分组的另一个常见用途是整理特定用户会话的所有活动事件，以找出用户进行的一系列操作（称为会话化（sessionization）【37】）。例如，可以使用这种分析来确定显示新版网站的用户是否比那些显示旧版本（A/B测试）的用户更有购买欲，或者计算某个营销活动是否值得。

如果你有多个Web服务器处理用户请求，则特定用户的活动事件很可能分散在各个不同的服务器的日志文件中。你可以通过使用会话cookie，用户ID或类似的标识符作为分组键，以将特定用户的所有活动事件放在一起来实现会话化，与此同时，不同用户的事件仍然散步在不同的分区中。

处理倾斜

如果存在与单个键关联的大量数据，则“将具有相同键的所有记录放到相同的位置”这种模式就被破坏了。例如在社交网络中，大多数用户可能会与几百人有连接，但少数名人可能有数百万的追随者。这种不成比例的活动数据库记录被称为关键对象（linchpin object）【38】或热键（hot key）。

在单个Reducer中收集与某个名流相关的所有活动（例如他们发布内容的回复）可能导致严重的倾斜（也称为热点（hot spot））—— 也就是说，一个Reducer必须比其他Reducer处理更多的记录（参见“负载倾斜与消除热点“）。由于MapReduce作业只有在所有Mapper和Reducer都完成时才完成，所有后续作业必须等待最慢的Reducer才能启动。

如果连接的输入存在热点键，可以使用一些算法进行补偿。例如，Pig中的倾斜连接（skewed join）方法首先运行一个抽样作业来确定哪些键是热键【39】。连接实际执行时，Mapper会将热键的关联记录随机（相对于传统MapReduce基于键散列的确定性方法）发送到几个Reducer之一。对于另外一侧的连接输入，与热键相关的记录需要被复制到所有处理该键的Reducer上【40】。

这种技术将处理热键的工作分散到多个Reducer上，这样可以使其更好地并行化，代价是需要将连接另一侧的输入记录复制到多个Reducer上。 Crunch中的**分片连接（sharded join）**方法与之类似，但需要显式指定热键而不是使用采样作业。这种技术也非常类似于我们在“负载倾斜与消除热点”中讨论的技术，使用随机化来缓解分区数据库中的热点。

Hive的偏斜连接优化采取了另一种方法。它需要在表格元数据中显式指定热键，并将与这些键相关的记录单独存放，与其它文件分开。当在该表上执行连接时，对于热键，它会使用Map端连接（参阅下一节）。

当按照热键进行分组并聚合时，可以将分组分两个阶段进行。第一个MapReduce阶段将记录发送到随机Reducer，以便每个Reducer只对热键的子集执行分组，为每个键输出一个更紧凑的中间聚合结果。然后第二个MapReduce作业将所有来自第一阶段Reducer的中间聚合结果合并为每个键一个值。

Map端连接

上一节描述的连接算法在Reducer中执行实际的连接逻辑，因此被称为Reduce端连接。Mapper扮演着预处理输入数据的角色：从每个输入记录中提取键值，将键值对分配给Reducer分区，并按键排序。

Reduce端方法的优点是不需要对输入数据做任何假设：无论其属性和结构如何，Mapper都可以对其预处理以备连接。然而不利的一面是，所有这些排序，复制至Reducer，以及合并Reducer输入可能开销巨大。当数据通过MapReduce 【37】阶段时，数据可能要落盘好几次，取决于可用的内存缓冲区。

另一方面，如果你能对输入数据作出某些假设，则通过使用所谓的Map端连接来加快连接速度是可行的。这种方法使用了一个阉掉Reduce与排序的MapReduce作业，每个Mapper只是简单地从分布式文件系统中读取一个输入文件块，然后将输出文件写入文件系统，仅此而已。

广播散列连接

适用于执行Map端连接的最简单场景是大数据集与小数据集连接的情况。要点在于小数据集需要足够小，以便可以将其全部加载到每个Mapper的内存中。

例如，假设在图10-2的情况下，用户数据库足够小以适应内存。在这种情况下，当Mapper启动时，它可以首先将用户数据库从分布式文件系统读取到内存中的哈希表中。完成此操作后，Map程序可以扫描用户活动事件，并简单地查找散列表中每个事件的用户标识⁷。

仍然可以有几个映射任务：一个用于连接的大输入的每个文件块（在图10-2的例子中，活动事件是大输入）。这些Mapper中的每一个都将小输入全部加载到内存中。

这种简单而有效的算法被称为广播散列连接：广播一词反映了这样一个事实，即大输入的分区的每个Mapper都读取整个小输入（所以小输入有效地“广播”到大的输入），单词hash反映了它使用一个哈希表。 Pig（名为“replicated join”），Hive（“MapJoin”），Cascading和Crunch支持此连接方法。它也用于数据仓库查询引擎，如Impala 【41】。

而不是将小连接输入加载到内存散列表中，另一种方法是将小连接输入存储在本地磁盘上的只读索引中【42】。该索引中经常使用的部分将保留在操作系统的页面缓存中，因此这种方法可以提供与内存中哈希表几乎一样快的随机访问查找，但实际上并不需要数据集适合内存。

分区散列连接

如果以相同方式对映射端连接的输入进行分区，则散列连接方法可以独立应用于每个分区。在图10-2的情况下，你可以根据用户标识的最后一位十进制数字来安排活动事件和用户数据库的每一个（因此每边有10个分区）。例如，Mapper3首先将所有具有以3结尾的ID的用户加载到散列表中，然后扫描ID为3的每个用户的所有活动事件。

如果分区正确完成，你可以确定所有你可能要加入的记录都位于相同编号的分区中，因此每个Mapper只能从每个输入数据集中读取一个分区就足够了。这具有的优点是每个Mapper都可以将较少量的数据加载到其哈希表中。

这种方法只适用于两个连接的输入具有相同数量的分区，记录根据相同的键和相同的散列函数分配给分区。如果输入是由之前执行过这个分组的MapReduce作业生成的，那么这可能是一个合理的假设。

分区散列连接在Hive 【37】中称为bucketed映射连接。Map端合并连接

如果输入数据集不仅以相同的方式进行分区，而且还基于相同的键进行排序，则应用另一种Map端联接的变体。在这种情况下，输入是否足够小以适应内存并不重要，因为Mapper可以执行通常由reducer执行的相同合并操作：按递增键递增读取两个输入文件，以及匹配相同的键记录。

如果Map端合并连接是可能的，则可能意味着先前的MapReduce作业首先将输入数据集引入到这个分区和排序的表单中。原则上，这个加入可以在之前工作的Reduce阶段进行。但是，在单独的仅用于Map的作业中执行合并连接仍然是适当的，例如，除了此特定连接之外，还需要分区和排序数据集以用于其他目的。

MapReduce与Map端连接的工作流程

当下游作业使用MapReduce连接的输出时，map-side或reduce-side连接的选择会影响输出的结构。 reduce-side连接的输出按连接键进行分区和排序，而map-side连接的输出按照与大输入相同的方式进行分区和排序（因为对每个文件块启动一个map任务无论是使用分区连接还是广播连接，连接的大输入）。

如前所述，Map边连接也对输入数据集的大小，排序和分区做出了更多的假设。在优化连接策略时，了解分布式文件系统中数据集的物理布局变得非常重要：仅仅知道编码格式和数据存储目录的名称是不够的;你还必须知道数据分区和排序的分区数量和键。

在Hadoop生态系统中，这种关于数据集分区的元数据经常在HCatalog和Hive Metastore中维护【37】。

工作流的输出

我们已经谈了很多关于实现MapReduce工作流程的各种算法，但是我们忽略了一个重要的问题：一旦完成，所有处理的结果是什么？我们为什么要把所有这些工作放在首位？

在数据库查询的情况下，我们根据分析目的来区分事务处理（OLTP）目的（参阅“事务处理或分析？”）。我们看到，OLTP查询通常使用索引按键查找少量记录，以便将其呈现给用户（例如，在网页上）。另一方面，分析查询通常会扫描大量记录，执行分组和汇总，输出通常具有报告的形式：显示某个指标随时间变化的图表，或前10个项目根据一些排名，或一些数量分解成子类别。这种报告的消费者通常是需要做出商业决策的分析师或经理。

批处理在哪里适合？这不是交易处理，也不是分析。与分析更接近，因为批处理过程通常扫描输入数据集的大部分。但是，MapReduce作业的工作流程与用于分析目的的SQL查询不同（参阅“比较Hadoop与分布式数据库”）。批处理过程的输出通常不是报告，而是一些其他类型的结构。

建立搜索索引

Google最初使用的MapReduce是为其搜索引擎建立索引，这个索引是作为5到10个MapReduce作业的工作流实现的【1】。虽然Google为了这个目的后来不再使用MapReduce 【43】，但是如果从建立搜索索引的角度来看，它可以帮助理解MapReduce。（即使在今天，Hadoop MapReduce仍然是构建Lucene / Solr索引的好方法。）

我们在“全文搜索和模糊索引”中简要地看到了Lucene这样的全文搜索索引是如何工作的：它是一个文件（关键词字典），你可以在其中高效地查找特定关键字并找到包含该关键字的所有文档ID列表（发布列表）。这是一个非常简单的搜索索引视图 —— 实际上，它需要各种附加数据，以便根据相关性对搜索结果进行排名，纠正拼写错误，解析同义词等等，但这一原则是成立的。

如果需要对一组固定文档执行全文搜索，则批处理是构建索引的一种非常有效的方法：Mapper根据需要对文档集进行分区，每个reducer构建其分区的索引，并将索引文件写入分布式文件系统。构建这样的文档分区索引（参阅“分区和二级索引”）并行处理非常好。

由于按关键字查询搜索索引是只读操作，因此这些索引文件一旦创建就是不可变的。

如果索引的文档集合发生更改，则可以选择定期重新运行整个索引工作流程，并在完成后用新的索引文件批量替换以前的索引文件。如果只有少量的文档发生了变化，这种方法可能会带来很高的计算成本，但是它的优点是索引过程很容易推理：文档，索引。

或者，可以逐渐建立索引。如第3章所述，如果要添加，删除或更新索引中的文档，Lucene会写出新的段文件，并异步合并和压缩背景中的段文件。我们将在第11章中看到更多这样的增量处理。

键值存储作为批处理输出

搜索索引只是批处理工作流程可能输出的一个示例。批量处理的另一个常见用途是构建机器学习系统，如分类器（例如，垃圾邮件过滤器，异常检测，图像识别）和推荐系统（例如，你可能认识的人，你可能感兴趣的产品或相关搜索）。

这些批处理作业的输出通常是某种数据库：例如，可以通过用户ID查询以获取该用户的建议朋友的数据库，或者可以通过产品ID查询的数据库以获取相关产品【45】。

这些数据库需要从处理用户请求的Web应用程序中查询，这些请求通常与Hadoop基础架构分离。那么批处理过程的输出如何返回到Web应用程序可以查询的数据库？

最明显的选择可能是直接在Mapper或Reducer中使用客户端库作为你最喜欢的数据库，并从批处理作业直接写入数据库服务器，一次写入一条记录。这将起作用（假设你的防火墙规则允许从你的Hadoop环境直接访问你的生产数据库），但由于以下几个原因，这是一个坏主意：

正如前面讨论的连接一样，为每个记录提出一个网络请求比批处理任务的正常吞吐量要慢几个数量级。即使客户端库支持批处理，性能也可能很差。
MapReduce作业经常并行运行许多任务。如果所有Mapper或Reducer都同时写入相同的输出数据库，并且批处理过程期望的速率，那么该数据库可能很容易被压倒，并且其查询性能可能受到影响。这可能会导致系统其他部分的操作问题【35】。
通常情况下，MapReduce为作业输出提供了一个干净的“全有或全无”的保证：如果作业成功，则结果就是只执行一次任务的输出，即使某些任务失败并且必须重试。如果整个作业失败，则不会生成输出。然而，从作业内部写入外部系统会产生外部可见的副作用，这种副作用是不能被隐藏的。因此，你不得不担心部分完成的作业对其他系统可见的结果，以及Hadoop任务尝试和推测性执行的复杂性。

更好的解决方案是在批处理作业中创建一个全新的数据库，并将其作为文件写入分布式文件系统中作业的输出目录，就像上一节的搜索索引一样。这些数据文件一旦写入就是不可变的，可以批量加载到处理只读查询的服务器中。各种键值存储支持在MapReduce作业中构建数据库文件，包括Voldemort 【46】，Terrapin 【47】，ElephantDB 【48】和HBase批量加载【49】。

构建这些数据库文件是MapReduce的一个很好的使用方法：使用Mapper提取一个键，然后使用该键进行排序已经成为构建索引所需的大量工作。由于大多数这些键值存储是只读的（文件只能由批处理作业一次写入，而且是不可变的），所以数据结构非常简单。例如，它们不需要WAL（参阅“使B树可靠”）。

将数据加载到Voldemort时，服务器将继续向旧数据文件提供请求，同时将新数据文件从分布式文件系统复制到服务器的本地磁盘。一旦复制完成，服务器会自动切换到查询新文件。如果在这个过程中出现任何问题，它可以很容易地再次切换回旧的文件，因为它们仍然存在，并且是不变的【46】。

批量过程输出的哲学

本章前面讨论过的Unix哲学（“Unix哲学”）鼓励通过对数据流的非常明确的实验来进行实验：程序读取输入并写入输出。在这个过程中，输入保持不变，任何以前的输出都被新输出完全替换，并且没有其他副作用。这意味着你可以随心所欲地重新运行一个命令，调整或调试它，而不会扰乱系统的状态。

MapReduce作业的输出处理遵循相同的原理。通过将输入视为不可变且避免副作用（如写入外部数据库），批处理作业不仅实现了良好的性能，而且更容易维护：

如果在代码中引入了一个错误，并且输出错误或损坏了，则可以简单地回滚到代码的先前版本，然后重新运行该作业，输出将再次正确。或者，甚至更简单，你可以将旧的输出保存在不同的目录中，然后切换回原来的目录。具有读写事务的数据库没有这个属性：如果你部署了错误的代码，将错误的数据写入数据库，那么回滚代码将无法修复数据库中的数据。（能够从错误代码中恢复的思想被称为人类容错【50】。）
由于易于回滚，功能开发可以比错误意味着不可挽回的损害的环境更快地进行。这种使不可逆性最小化的原则有利于敏捷软件的开发【51】。
如果映射或减少任务失败，MapReduce框架将自动重新调度并在同一个输入上再次运行它。如果失败是由于代码中的一个错误造成的，那么它会一直崩溃，并最终导致作业在几次尝试之后失败。但是如果故障是由于暂时的问题引起的，那么故障是可以容忍的。这种自动重试只是安全的，因为输入是不可变的，而失败任务的输出被MapReduce框架丢弃。
同一组文件可用作各种不同作业的输入，其中包括计算度量标准的计算作业，并评估作业的输出是否具有预期的特性（例如，将其与前一次运行的输出进行比较并测量差异）。
与Unix工具类似，MapReduce作业将逻辑与布线（配置输入和输出目录）分开，这就提供了关注点的分离，并且可以重用代码：一个团队可以专注于实现一件好事的工作其他团队可以决定何时何地运行这项工作。

在这些领域，对Unix运行良好的设计原则似乎也适用于Hadoop，但Unix和Hadoop在某些方面也有所不同。例如，因为大多数Unix工具都假定没有类型的文本文件，所以他们必须做大量的输入解析（本章开头的日志分析示例使用{print $7}来提取URL）。在Hadoop上，通过使用更多结构化的文件格式，可以消除一些低价值的语法转换：Avro（参阅“Avro”）和Parquet（参阅第95页上的“列存储”）经常使用，因为它们提供高效的基于模式的编码，并允许随着时间的推移模式的演变（见第4章）。

比较Hadoop和分布式数据库

正如我们所看到的，Hadoop有点像Unix的分布式版本，其中HDFS是文件系统，而MapReduce是Unix进程的古怪实现（这恰好总是在映射阶段和缩小阶段之间运行排序实用程序）。我们看到了如何在这些基元之上实现各种连接和分组操作。

当MapReduce论文【1】发表时，它在某种意义上说并不新鲜。我们在前几节中讨论的所有处理和并行连接算法已经在十多年前的所谓的**大规模并行处理（MPP， massively parallel processing）**数据库中实现了【3,40】。例如，Gamma数据库机器，Teradata和Tandem NonStop SQL是这方面的先驱【52】。

最大的区别是MPP数据库集中于在一组机器上并行执行分析SQL查询，而MapReduce和分布式文件系统【19】的组合则更像是一个可以运行任意程序的通用操作系统。

存储的多样性

数据库要求你根据特定的模型（例如关系或文档）来构造数据，而分布式文件系统中的文件只是字节序列，可以使用任何数据模型和编码来编写。它们可能是数据库记录的集合，但同样可以是文本，图像，视频，传感器读数，稀疏矩阵，特征向量，基因组序列或任何其他类型的数据。

说白了，Hadoop开放了将数据不加区分地转储到HDFS的可能性，之后才想出如何进一步处理它【53】。相比之下，在将数据导入数据库专有存储格式之前，MPP数据库通常需要对数据和查询模式进行仔细的前期建模。

从纯粹的角度来看，这种仔细的建模和导入似乎是可取的，因为这意味着数据库的用户有更好的质量数据来处理。然而，在实践中，似乎只是简单地使数据可用 —— 即使它是一个古怪的，难以使用的原始格式 —— 通常比尝试决定理想的数据模型更有价值[54 ]。

这个想法与数据仓库类似（参阅“数据仓库”）：将大型组织的各个部分的数据集中在一起是很有价值的，因为它可以跨以前不同的数据集进行联接。 MPP数据库所要求的谨慎的模式设计减慢了集中式数据收集速度；以原始形式收集数据，以后担心模式设计，使数据收集速度加快（有时被称为“数据湖（data lake）”或“企业数据中心（enterprise data hub）”【55】）。

不加区别的数据倾销改变了解释数据的负担：不是强迫数据集的生产者将其转化为标准化的格式，而是数据的解释成为消费者的问题（读时模式方法【56】；参阅“文档模型中的架构灵活性”）。如果生产者和消费者是不同优先级的不同团队，这可能是一个优势。甚至可能不存在一个理想的数据模型，而是对适合不同目的的数据有不同的看法。以原始形式简单地转储数据可以进行多次这样的转换。这种方法被称为寿司原则：“原始数据更好”【57】。

因此，Hadoop经常被用于实现ETL过程（参阅“数据仓库”）：事务处理系统中的数据以某种原始形式转储到分布式文件系统中，然后编写MapReduce作业来清理数据，将其转换为关系表单，并将其导入MPP数据仓库以进行分析。数据建模仍然在发生，但它是在一个单独的步骤中，从数据收集中分离出来的。这种解耦是可能的，因为分布式文件系统支持以任何格式编码的数据。

加工模型的多样性

MPP数据库是单一的，紧密集成的软件，负责磁盘上的存储布局，查询计划，调度和执行。由于这些组件都可以针对数据库的特定需求进行调整和优化，因此整个系统可以在其设计的查询类型上取得非常好的性能。而且，SQL查询语言允许表达式查询和重要语义，而无需编写代码，使业务分析师（例如Tableau）使用的图形工具可访问该语言。

另一方面，并非所有类型的处理都可以合理地表达为SQL查询。例如，如果要构建机器学习和推荐系统，或者使用相关性排名模型的全文搜索索引，或者执行图像分析，则很可能需要更一般的数据处理模型。这些类型的处理通常对特定的应用程序非常具体（例如机器学习的特征工程，机器翻译的自然语言模型，欺诈预测的风险评估函数），因此它们不可避免地需要编写代码，而不仅仅是查询。

MapReduce使工程师能够轻松地在大型数据集上运行自己的代码。如果你有HDFS和MapReduce，那么你可以在它上面建立一个SQL查询执行引擎，事实上这正是Hive项目所做的【31】。但是，你也可以编写许多其他形式的批处理，这些批处理不适合用SQL查询表示。

随后，人们发现MapReduce对于某些类型的处理来说太过于限制，执行得太差，因此其他各种处理模型都是在Hadoop之上开发的（我们将在“后MapReduce时代”中看到其中的一些）。有两种处理模型，SQL和MapReduce，还不够，需要更多不同的模型！而且由于Hadoop平台的开放性，实施一整套方法是可行的，而这在整体MPP数据库的范围内是不可能的【58】。

至关重要的是，这些不同的处理模型都可以在一个共享的机器上运行，所有这些机器都可以访问分布式文件系统上的相同文件。在Hadoop方法中，不需要将数据导入到几个不同的专用系统中进行不同类型的处理：系统足够灵活，可以支持同一个群集内不同的工作负载。不需要移动数据使得从数据中获得价值变得容易得多，并且使用新的处理模型更容易进行实验。

Hadoop生态系统包括随机访问的OLTP数据库，如HBase（参阅“SSTables和LSM树”）和MPA样式的分析数据库，如Impala 【41】。 HBase和Impala都不使用MapReduce，但都使用HDFS进行存储。它们是访问和处理数据的非常不同的方法，但是它们可以共存并被集成到同一个系统中。

为频繁的故障而设计

在比较MapReduce和MPP数据库时，设计方法的另外两个不同点是：处理故障和使用内存和磁盘。与在线系统相比，批处理对故障不太敏感，因为如果失败，用户不会立即影响用户，并且可以再次运行。

如果一个节点在执行查询时崩溃，大多数MPP数据库会中止整个查询，并让用户重新提交查询或自动重新运行它【3】。由于查询通常最多运行几秒钟或几分钟，所以这种处理错误的方法是可以接受的，因为重试的代价不是太大。 MPP数据库还倾向于在内存中保留尽可能多的数据（例如，使用散列连接）以避免从磁盘读取的成本。

另一方面，MapReduce可以容忍映射或减少任务的失败，而不会影响作业的整体，通过以单个任务的粒度重试工作。它也非常渴望将数据写入磁盘，一方面是为了容错，另一方面是假设数据集太大而不能适应内存。

MapReduce方法更适用于较大的作业：处理如此之多的数据并运行很长时间的作业，以至于在此过程中可能至少遇到一个任务故障。在这种情况下，由于单个任务失败而重新运行整个工作将是浪费的。即使以单个任务的粒度进行恢复引入了使得无故障处理更慢的开销，但如果任务失败率足够高，仍然可以进行合理的权衡。

但是这些假设有多现实呢？在大多数集群中，机器故障确实发生，但是它们不是很频繁 —— 可能很少，大多数工作都不会经验，因为机器故障。为了容错，真的值得引起重大的开销吗？

要了解MapReduce节省使用内存和任务级恢复的原因，查看最初设计MapReduce的环境是很有帮助的。 Google拥有混合使用的数据中心，在线生产服务和离线批处理作业在同一台机器上运行。每个任务都有一个使用容器执行的资源分配（CPU核心，RAM，磁盘空间等）。每个任务也具有优先级，如果优先级较高的任务需要更多的资源，则可以终止（抢占）同一台机器上较低优先级的任务以释放资源。优先级还决定了计算资源的定价：团队必须为他们使用的资源付费，而优先级更高的流程花费更多【59】。

这种架构允许非生产（低优先级）计算资源被过度使用，因为系统知道如果必要的话它可以回收资源。与分离生产和非生产任务的系统相比，过度使用资源可以更好地利用机器和提高效率。但是，由于MapReduce作业以低优先级运行，因此它们随时都有被抢占的风险，因为优先级较高的进程需要其资源。批量工作有效地“拿起桌子下面的碎片”，利用高优先级进程已经采取的任何计算资源。

在谷歌，运行一个小时的MapReduce任务有大约5％被终止的风险，为更高优先级的进程腾出空间。由于硬件问题，机器重新启动或其他原因，这个速率比故障率高出一个数量级【59】。按照这种抢先率，如果一个作业有100个任务，每个任务运行10分钟，那么至少有一个任务在完成之前将被终止的风险大于50％。

这就是为什么MapReduce能够容忍频繁意外的任务终止的原因：这不是因为硬件特别不可靠，这是因为任意终止进程的自由可以在计算集群中更好地利用资源。

在开源的集群调度器中，抢占的使用较少。 YARN的CapacityScheduler支持抢占以平衡不同队列的资源分配【58】，但在编写本文时，YARN，Mesos或Kubernetes不支持通用优先级抢占【60】。在任务不经常被终止的环境中，MapReduce的设计决策没有多少意义。在下一节中，我们将看看MapReduce的一些替代方案，这些替代方案做出了不同的设计决定。

后MapReduce时代

虽然MapReduce在二十世纪二十年代后期变得非常流行并受到大量的炒作，但它只是分布式系统的许多可能的编程模型之一。根据数据量，数据结构和处理类型，其他工具可能更适合表达计算。

尽管如此，我们在讨论MapReduce的这一章花了很多时间，因为它是一个有用的学习工具，因为它是分布式文件系统的一个相当清晰和简单的抽象。也就是说，能够理解它在做什么，而不是在易于使用的意义上是简单的。恰恰相反：使用原始的MapReduce API来实现复杂的处理工作实际上是非常困难和费力的 —— 例如，你需要从头开始实现任何连接算法【37】。

针对直接使用MapReduce的困难，在MapReduce上创建了各种更高级的编程模型（Pig，Hive，Cascading，Crunch）作为抽象。如果你了解MapReduce的工作原理，那么它们相当容易学习，而且它们的高级构造使许多常见的批处理任务更容易实现。

但是，MapReduce执行模型本身也存在一些问题，这些问题并没有通过增加另一个抽象层次来解决，而且在某些类型的处理中表现得很差。一方面，MapReduce非常强大：你可以使用它来处理频繁任务终止的不可靠多租户系统上几乎任意大量的数据，并且仍然可以完成工作（虽然速度很慢）。另一方面，对于某些类型的处理来说，其他工具有时也会更快。

在本章的其余部分中，我们将介绍一些批处理方法。在第十一章我们将转向流处理，这可以看作是加速批处理的另一种方法。

内部状态表示

如前所述，每个MapReduce作业都独立于其他任何作业。作业与世界其他地方的主要联系点是分布式文件系统上的输入和输出目录。如果希望一个作业的输出成为第二个作业的输入，则需要将第二个作业的输入目录配置为与第一个作业的输出目录相同，并且外部工作流调度程序必须仅在第一份工作已经完成。

如果第一个作业的输出是要在组织内广泛发布的数据集，则此设置是合理的。在这种情况下，你需要能够通过名称来引用它，并将其用作多个不同作业（包括由其他团队开发的作业）的输入。将数据发布到分布式文件系统中的众所周知的位置允许松耦合，这样作业就不需要知道是谁在输入输出或消耗其输出（参阅“分离逻辑和布线”在本页395）。

但是，在很多情况下，你知道一个工作的输出只能用作另一个工作的输入，这个工作由同一个团队维护。在这种情况下，分布式文件系统上的文件只是简单的中间状态：一种将数据从一个作业传递到下一个作业的方式。在用于构建由50或100个MapReduce作业【29】组成的推荐系统的复杂工作流程中，存在很多这样的中间状态。

将这个中间状态写入文件的过程称为物化。（我们在“聚合：数据立方体和物化视图”中已经在物化视图的背景下遇到了这个术语。它意味着要急于计算某个操作的结果并写出来，而不是计算需要时按要求。）

相反，本章开头的日志分析示例使用Unix管道将一个命令的输出与另一个命令的输出连接起来。管道并没有完全实现中间状态，而是只使用一个小的内存缓冲区，将输出逐渐流向输入。

MapReduce的完全实现中间状态的方法与Unix管道相比存在不足：

MapReduce作业只有在前面的作业（生成其输入）中的所有任务都完成时才能启动，而由Unix管道连接的进程同时启动，输出一旦生成就会被使用。不同机器上的偏差或不同的负荷意味着一份工作往往会有一些比其他人更快完成的离散任务。必须等到所有前面的工作完成才能减慢整个工作流程的执行。
Mapper通常是多余的：它们只读取刚刚由reducer写入的相同文件，并为下一个分区和排序阶段做好准备。在许多情况下，Mapper代码可能是以前的reducer的一部分：如果reducer输出被分区和排序的方式与mapper输出相同，那么reducers可以直接链接在一起，而不与mapper阶段交错。
将中间状态存储在分布式文件系统中意味着这些文件被复制到多个节点，这对于这样的临时数据通常是过度的。

数据流引擎

了解决MapReduce的这些问题，开发了几种用于分布式批量计算的新的执行引擎，其中最着名的是Spark 【61,62】，Tez 【63,64】和Flink 【65,66】。他们设计的方式有很多不同之处，但他们有一个共同点：他们把整个工作流作为一项工作来处理，而不是把它分解成独立的子作业。

由于它们通过几个处理阶段明确地建模数据流，所以这些系统被称为数据流引擎。像MapReduce一样，它们通过反复调用用户定义的函数来在单个线程上一次处理一条记录。他们通过对输入进行分区来并行工作，并将一个功能的输出复制到网络上，成为另一个功能的输入。

与MapReduce不同，这些功能不需要交替映射和Reduce的严格角色，而是可以以更灵活的方式进行组合。我们称之为这些函数操作符，数据流引擎提供了几个不同的选项来连接一个操作符的输出到另一个的输入：

一个选项是通过键对记录进行重新分区和排序，就像在MapReduce的混洗阶段一样（参阅“分布式执行MapReduce”）。此功能可以像在MapReduce中一样启用排序合并连接和分组。
另一种可能是采取几个输入，并以相同的方式进行分区，但跳过排序。这节省了分区散列连接的工作，其中记录的分区是重要的，但顺序是不相关的，因为构建散列表随机化了顺序。
对于广播散列连接，可以将一个运算符的相同输出发送到连接运算符的所有分区。

这种处理引擎的风格基于像Dryad 【67】和Nephele 【68】这样的研究系统，与MapReduce模型相比，它提供了几个优点：

排序等昂贵的工作只需要在实际需要的地方执行，而不是在每个Map和Reduce阶段之间默认发生。
没有不必要的Map任务，因为Mapper所做的工作通常可以合并到前面的reduce操作器中（因为Mapper不会更改数据集的分区）。
由于工作流程中的所有连接和数据依赖性都是明确声明的，因此调度程序会概述哪些数据是必需的，因此可以进行本地优化。例如，它可以尝试将占用某些数据的任务放在与生成它的任务相同的机器上，以便可以通过共享内存缓冲区交换数据，而不必通过网络复制数据。
通常将操作员之间的中间状态保存在内存中或写入本地磁盘就足够了，这比写入HDFS需要更少的I/O（必须将其复制到多个计算机并写入到每个代理的磁盘上）。 MapReduce已经将这种优化用于Mapper的输出，但是数据流引擎将该思想推广到了所有的中间状态。
操作员可以在输入准备就绪后立即开始执行;在下一个开始之前不需要等待整个前一阶段的完成。
与MapReduce（为每个任务启动一个新的JVM）相比，现有的Java虚拟机（JVM）进程可以重用来运行新操作，从而减少启动开销。

你可以使用数据流引擎来执行与MapReduce工作流相同的计算，并且由于此处所述的优化，通常执行速度会明显更快。既然操作符是map和reduce的泛化，相同的处理代码可以在任一执行引擎上运行：Pig，Hive或Cascading中实现的工作流可以通过简单的配置更改从MapReduce切换到Tez或Spark，而无需修改代码【64】。

Tez是一个相当薄的库，它依赖于YARN shuffle服务来实现节点间数据的实际复制【58】，而Spark和Flink则是包含自己的网络通信层，调度器和面向用户的API的大型框架。我们将在短期内讨论这些高级API。

容错

完全实现中间状态到分布式文件系统的一个优点是它是持久的，这使得MapReduce中的容错相当容易：如果一个任务失败，它可以在另一台机器上重新启动，并从文件系统重新读取相同的输入。

Spark，Flink和Tez避免将中间状态写入HDFS，因此他们采取了不同的方法来容忍错误：如果一台机器发生故障，并且该机器上的中间状态丢失，则会从其他仍然可用的数据重新计算在可能的情况下是在先的中间阶段，或者是通常在HDFS上的原始输入数据）。

为了实现这个重新计算，框架必须跟踪一个给定的数据是如何计算的 —— 使用哪个输入分区，以及哪个操作符被应用到它。 Spark使用弹性分布式数据集（RDD）抽象来追踪数据的祖先【61】，而Flink检查点操作符状态，允许其恢复运行在执行过程中遇到错误的操作符【66】。

在重新计算数据时，重要的是要知道计算是否是确定性的：也就是说，给定相同的输入数据，操作员是否始终生成相同的输出？如果一些丢失的数据已经发送给下游运营商，这个问题就很重要。如果运营商重新启动，重新计算的数据与原有的丢失数据不一致，下游运营商很难解决新旧数据之间的矛盾。对于不确定性运营商来说，解决方案通常是杀死下游运营商，然后再运行新数据。

为了避免这种级联故障，最好让运算符具有确定性。但是请注意，非确定性行为很容易发生意外蔓延：例如，许多编程语言在迭代哈希表的元素时不能保证任何特定顺序，许多概率和统计算法明确依赖于使用随机数，以及任何用途系统时钟或外部数据源是不确定的。为了可靠地从故障中恢复，例如通过使用固定种子产生伪随机数，需要消除这种不确定性的原因。

通过重新计算数据从故障中恢复并不总是正确的答案：如果中间数据比源数据小得多，或者如果计算量非常大，那么将中间数据转化为文件可能比将其重新计算更便宜。

关于物化的讨论

回到Unix的类比，我们看到MapReduce就像是将每个命令的输出写入临时文件，而数据流引擎看起来更像是Unix管道。尤其是Flink是围绕流水线执行的思想而建立的：也就是说，将运算符的输出递增地传递给其他操作符，并且在开始处理之前不等待输入完成。

排序操作不可避免地需要消耗其整个输入，然后才能生成任何输出，因为最后一个输入记录可能是具有最低键的输入记录，因此需要作为第一个输出记录。任何需要分类的操作员都需要至少暂时地累积状态。但是工作流程的许多其他部分可以以流水线方式执行。

当作业完成时，它的输出需要持续到某个地方，以便用户可以找到并使用它—— 很可能它会再次写入分布式文件系统。因此，在使用数据流引擎时，HDFS上的物化数据集通常仍是作业的输入和最终输出。和MapReduce一样，输入是不可变的，输出被完全替换。对MapReduce的改进是，你可以节省自己将所有中间状态写入文件系统。

图与迭代处理

在“图数据模型”中，我们讨论了使用图形来建模数据，并使用图形查询语言来遍历图形中的边和顶点。第2章的讨论集中在OLTP风格的使用上：快速执行查询来查找少量符合特定条件的顶点。

在批处理环境中查看图也很有趣，其目标是在整个图上执行某种离线处理或分析。这种需求经常出现在机器学习应用程序（如推荐引擎）或排序系统中。例如，最着名的图形分析算法之一是PageRank 【69】，它试图根据其他网页链接的网页来估计网页的流行度。它被用作确定网络搜索引擎呈现结果的顺序的公式的一部分。

像Spark，Flink和Tez这样的数据流引擎（参见“中间状态的物化”）通常将操作符作为**有向无环图（DAG）**排列在作业中。这与图形处理不一样：在数据流引擎中，从一个操作符到另一个操作符的数据流被构造成一个图，而数据本身通常由关系式元组构成。在图形处理中，数据本身具有图形的形式。另一个不幸的命名混乱！

许多图算法是通过一次遍历一个边来表示的，将一个顶点与相邻的顶点连接起来以便传播一些信息，并且重复直到满足一些条件为止——例如，直到没有更多的边要跟随，或者直到一些度量收敛。我们在图2-6中看到一个例子，它通过重复地跟踪指示哪个位置在哪个其他位置（这种算法被称为传递闭包）的边缘，列出了包含在数据库中的北美所有位置。

可以在分布式文件系统（包含顶点和边的列表的文件）中存储图形，但是这种“重复直到完成”的想法不能用普通的MapReduce来表示，因为它只执行一次数据传递。这种算法因此经常以迭代方式实现：

外部调度程序运行批处理来计算算法的一个步骤。
当批处理过程完成时，调度器检查它是否完成（基于完成条件 - 例如，没有更多的边要跟随，或者与上次迭代相比的变化低于某个阈值）。
如果尚未完成，则调度程序返回到步骤1并运行另一轮批处理。

这种方法是有效的，但是用MapReduce实现它往往是非常低效的，因为MapReduce没有考虑算法的迭代性质：它总是读取整个输入数据集并产生一个全新的输出数据集，即使只有一小部分该图与上次迭代相比已经改变。 Pregel处理模型

作为批处理图形的优化，计算的批量同步并行（BSP）模型【70】已经流行起来。其中，它由Apache Giraph 【37】，Spark的GraphX API和Flink的Gelly API 【71】实现。它也被称为Pregel模型，正如Google的Pregel论文推广这种处理图的方法【72】。

回想一下在MapReduce中，Mapper在概念上“发送消息”给reducer的特定调用，因为框架将所有的mapper输出集中在一起。 Pregel背后有一个类似的想法：一个顶点可以“发送消息”到另一个顶点，通常这些消息沿着图的边被发送。

在每次迭代中，为每个顶点调用一个函数，将所有发送给它的消息传递给它 —— 就像调用reducer一样。与MapReduce的不同之处在于，在Pregel模型中，顶点从一次迭代到下一次迭代记忆它的状态，所以这个函数只需要处理新的传入消息。如果在图的某个部分没有发送消息，则不需要做任何工作。

这与演员模型有些相似（参阅“分布式的Actor框架”），除非顶点状态和顶点之间的消息具有容错性和耐久性，并且通信以固定的方式进行，否则将每个顶点视为主角轮次：在每一次迭代中，框架传递在前一次迭代中发送的所有消息。演员通常没有这样的时间保证。

容错

顶点只能通过消息传递进行通信（而不是直接相互查询）的事实有助于提高Pregel作业的性能，因为消息可以成批处理，而且等待通信的次数也减少了。唯一的等待是在迭代之间：由于Pregel模型保证所有在一次迭代中发送的消息都在下一次迭代中传递，所以先前的迭代必须完全完成，并且所有的消息必须在网络上复制，然后下一个开始。即使底层网络可能丢失，重复或任意延迟消息（参阅“不可靠的网络”），Pregel实施可保证在接下来的迭代中消息在其目标顶点处理一次。像MapReduce一样，该框架透明地从故障中恢复，以简化Pregel顶层算法的编程模型。

这种容错是通过在迭代结束时定期检查所有顶点的状态来实现的，即将其全部状态写入持久存储。如果某个节点发生故障并且其内存中状态丢失，则最简单的解决方法是将整个图计算回滚到上一个检查点，然后重新启动计算。如果算法是确定性的并且记录了消息，那么也可以选择性地只恢复丢失的分区（就像我们之前讨论过的数据流引擎）【72】。

并行执行

顶点不需要知道它正在执行哪个物理机器;当它发送消息到其他顶点时，它只是将它们发送到一个顶点ID。分配图的框架，即确定哪个顶点运行在哪个机器上，以及如何通过网络路由消息，以便它们结束在正确的位置。

由于编程模型一次仅处理一个顶点（有时称为“像顶点一样思考”），所以框架可以以任意方式划分图形。理想情况下，如果它们需要进行大量的通信，那么它将被分割，以使顶点在同一台机器上共置。然而，寻找这样一个优化的分割在实践中是困难的，图形经常被任意分配的顶点ID分割，而不会尝试将相关的顶点分组在一起。

因此，图算法通常会有很多跨机器通信，而中间状态（节点之间发送的消息）往往比原始图大。通过网络发送消息的开销会显着减慢分布式图算法的速度。

出于这个原因，如果你的图可以放在一台计算机的内存中，那么单机（甚至可能是单线程）算法很可能会超越分布式批处理【73,74】。即使图大于内存，也可以放在单个计算机的磁盘上，使用GraphChi等框架进行单机处理是一个可行的选择【75】。如果图太大而不适合单个机器，像Pregel这样的分布式方法是不可避免的。有效的并行化图算法是一个正在进行的领域。

高级API和语言

自MapReduce第一次流行以来，分布式批处理的执行引擎已经成熟。到目前为止，基础设施已经足够强大，能够存储和处理超过10,000台机器群集上的数PB的数据。由于在这种规模下物理操作批处理过程的问题已经或多或少得到了解决，所以已经转向其他领域：改进编程模型，提高处理效率，扩大这些技术可以解决的问题集。

如前所述，Hive，Pig，Cascading和Crunch等高级语言和API由于手工编写MapReduce作业而变得非常流行。随着Tez的出现，这些高级语言还有额外的好处，可以移动到新的数据流执行引擎，而无需重写作业代码。 Spark和Flink也包括他们自己的高级数据流API，经常从FlumeJava中获得灵感【34】。

这些数据流API通常使用关系式构建块来表达一个计算：连接数据集以获取某个字段的值;按键分组元组;过滤一些条件;并通过计数，求和或其他函数来聚合元组。在内部，这些操作是使用本章前面讨论过的各种连接和分组算法来实现的。

除了需要较少代码的明显优势之外，这些高级接口还允许交互式使用，在这种交互式使用中，你可以将分析代码逐步编写到shell中并经常运行，以观察它正在做什么。这种发展风格在探索数据集和试验处理方法时非常有用。这也让人联想到Unix哲学，我们在第394页的“Unix哲学”中讨论过这个问题。

而且，这些高级接口不仅使人类使用系统的效率更高，而且提高了机器级别的工作执行效率。

向声明式查询语言的转变

与拼写执行连接的代码相比，指定连接为关系运算符的优点是，框架可以分析连接输入的属性，并自动决定哪个上述连接算法最适合手头的任务。 Hive，Spark和Flink都有基于代价的查询优化器，可以做到这一点，甚至可以改变连接顺序，使中间状态的数量最小化【66,77,78,79】。

连接算法的选择可以对批处理作业的性能产生很大的影响，不必理解和记住本章中讨论的各种连接算法。如果以声明的方式指定连接，则这是可能的：应用程序简单地说明哪些连接是必需的，查询优化器决定如何最好地执行连接。我们以前在第42页的“数据的查询语言”中遇到了这个想法。

但是，在其他方面，MapReduce及其数据流后继与SQL的完全声明性查询模型有很大不同。 MapReduce是围绕回调函数的思想构建的：对于每个记录或者一组记录，调用一个用户定义的函数（Mapper或Reducer），并且该函数可以自由地调用任意代码来决定输出什么。这种方法的优点是可以绘制在现有库的大型生态系统上进行分析，自然语言分析，图像分析以及运行数字或统计算法等。

轻松运行任意代码的自由是从MPP数据库（参见“比较Hadoop和分布式数据库”一节）中分离出来的MapReduce传统批处理系统。虽然数据库具有编写用户定义函数的功能，但是它们通常使用起来很麻烦，而且与大多数编程语言中广泛使用的程序包管理器和依赖管理系统（例如Maven for Java，npm for Java-Script，和Ruby的Ruby的Ruby）。

但是，数据流引擎已经发现，除了连接之外，在合并更多的声明性特征方面也是有优势的。例如，如果一个回调函数只包含一个简单的过滤条件，或者只是从一条记录中选择了一些字段，那么在调用每条记录的函数时会有相当大的CPU开销。如果以声明方式表示这样简单的过滤和映射操作，那么查询优化器可以利用面向列的存储布局（参阅“列存储”），并从磁盘只读取所需的列。 Hive，Spark DataFrames和Impala也使用向量化执行（参阅“内存带宽和向量处理”）：在对CPU缓存很友好的内部循环中迭代数据，并避免函数调用。

Spark生成JVM字节码【79】，Impala使用LLVM为这些内部循环生成本机代码【41】。

通过将声明性方面与高级API结合起来，并使查询优化器可以在执行期间利用这些优化方法，批处理框架看起来更像MPP数据库（并且可以实现可比较的性能）。同时，通过具有运行任意代码和以任意格式读取数据的可扩展性，它们保持了灵活性的优势。

专业化的不同领域

尽管能够运行任意代码的可扩展性是有用的，但是在标准处理模式不断重复发生的情况下也有许多常见的情况，所以值得重用通用构建块的实现。传统上，MPP数据库满足了商业智能分析师和业务报告的需求，但这只是许多使用批处理的领域之一。

另一个越来越重要的领域是统计和数值算法，它们是机器学习应用（如分类和推荐系统）所需要的。可重复使用的实现正在出现：例如，Mahout在MapReduce，Spark和Flink之上实现了用于机器学习的各种算法，而MADlib在关系型MPP数据库（Apache HAWQ）中实现了类似的功能【54】。

空间算法也是有用的，例如最近邻搜索（kNN）【80】，它在一些多维空间中搜索与给定物品接近的物品 - 这是一种类似的搜索。近似搜索对于基因组分析算法也很重要，它们需要找到相似但不相同的字符串【81】。

批处理引擎正被用于分布式执行来自日益广泛的领域的算法。随着批处理系统获得内置功能和高级声明性操作符，并且随着MPP数据库变得更加可编程和灵活，两者开始看起来更相似：最终，它们都只是存储和处理数据的系统。

本章小结

在本章中，我们探讨了批处理的主题。我们首先查看了诸如awk，grep和sort之类的Unix工具，然后我们看到了这些工具的设计理念是如何运用到MapReduce和更新的数据流引擎中的。这些设计原则中的一些是输入是不可变的，输出是为了成为另一个（还未知的）程序的输入，而复杂的问题是通过编写“做一件好事”的小工具来解决的。

在Unix世界中，允许一个程序与另一个程序组合的统一界面是文件和管道；在MapReduce中，该接口是一个分布式文件系统。我们看到数据流引擎添加了自己的管道式数据传输机制，以避免将中间状态转化为分布式文件系统，但作业的初始输入和最终输出通常仍然是HDFS。

分布式批处理框架需要解决的两个主要问题是：

分区

在MapReduce中，Mapper根据输入文件块进行分区。Mapper的输出被重新分区，分类并合并到可配置数量的Reducer分区中。这个过程的目的是把所有的相关数据 —— 例如，所有的记录都放在同一个地方。

后MapReduce数据流引擎尽量避免排序，除非它是必需的，但它们采取了大致类似的分区方法。

容错

MapReduce经常写入磁盘，这使得从单个失败的任务中轻松地恢复，而无需重新启动整个作业，但在无故障的情况下减慢了执行速度。数据流引擎对中间状态执行的实现较少，并且保留在内存中，这意味着如果节点发生故障，则需要推荐更多的数据。确定性运算符减少了需要重新计算的数据量。

我们讨论了几种MapReduce的连接算法，其中大多数也是在MPP数据库和数据流引擎中使用的。他们还提供了分区算法如何工作的一个很好的例子：

排序合并连接

每个正在连接的输入都通过一个提取连接键的Mapper。通过分区，排序和合并，具有相同键的所有记录最终都会进入Reducer的相同调用。这个函数可以输出连接的记录。

广播散列连接

两个连接输入之一是小的，所以它没有分区，它可以被完全加载到一个哈希表。因此，你可以为大连接输入的每个分区启动一个Mapper，将小输入的散列表加载到每个Mapper中，然后一次扫描大输入一条记录，查询每条记录的散列表。

分区散列连接

如果两个连接输入以相同的方式分区（使用相同的键，相同的散列函数和相同数量的分区），则可以独立地为每个分区使用散列表方法。

分布式批处理引擎有一个刻意限制的编程模型：回调函数（比如Mapper和Reducer）被认为是无状态的，除了指定的输出外，没有外部可见的副作用。这个限制允许框架隐藏抽象背后的一些困难的分布式系统问题：面对崩溃和网络问题，任务可以安全地重试，任何失败任务的输出都被丢弃。如果某个分区的多个任务成功，则只有其中一个实际上使其输出可见。

得益于这个框架，你在批处理作业中的代码无需担心实现容错机制：框架可以保证作业的最终输出与没有发生错误的情况相同，也许不得不重新尝试各种任务。这些可靠的语义要比在线服务处理用户请求时经常使用的要多得多，而且在处理请求的副作用时写入数据库。

批量处理工作的显着特点是它读取一些输入数据并产生一些输出数据，而不修改输入—— 换句话说，输出是从输入衍生出的。重要的是，输入数据是有界的：它有一个已知的，固定的大小（例如，它包含一些时间点的日志文件或数据库内容的快照）。因为它是有界的，一个工作知道什么时候它完成了整个输入的读取，所以一个工作最终完成。

在下一章中，我们将转向流处理，其中的输入是未知的 —— 也就是说，你还有一份工作，但是它的输入是永无止境的数据流。在这种情况下，工作永远不会完成，因为在任何时候都可能有更多的工作进来。我们将看到流和批处理在某些方面是相似的。但是关于无尽数据流的假设，也对我们构建系统的方式产生了很大的改变。

参考文献

Jeffrey Dean and Sanjay Ghemawat: “MapReduce: Simplified Data Processing on Large Clusters,” at 6th USENIX Symposium on Operating System Design and Implementation (OSDI), December 2004.
Joel Spolsky: “The Perils of JavaSchools,” joelonsoftware.com, December 25, 2005.
Shivnath Babu and Herodotos Herodotou: “Massively Parallel Databases and MapReduce Systems,” Foundations and Trends in Databases, volume 5, number 1, pages 1–104, November 2013. doi:10.1561/1900000036
David J. DeWitt and Michael Stonebraker: “MapReduce: A Major Step Backwards,” originally published at databasecolumn.vertica.com, January 17, 2008.
Henry Robinson: “The Elephant Was a Trojan Horse: On the Death of Map-Reduce at Google,” the-paper-trail.org, June 25, 2014.
“The Hollerith Machine,” United States Census Bureau, census.gov.
“IBM 82, 83, and 84 Sorters Reference Manual,” Edition A24-1034-1, International Business Machines Corporation, July 1962.
Adam Drake: “Command-Line Tools Can Be 235x Faster than Your Hadoop Cluster,” aadrake.com, January 25, 2014.
“GNU Coreutils 8.23 Documentation,” Free Software Foundation, Inc., 2014.
Martin Kleppmann: “Kafka, Samza, and the Unix Philosophy of Distributed Data,” martin.kleppmann.com, August 5, 2015.
Doug McIlroy:Internal Bell Labs memo, October 1964. Cited in: Dennis M. Richie: “Advice from Doug McIlroy,” cm.bell-labs.com.
M. D. McIlroy, E. N. Pinson, and B. A. Tague: “UNIX Time-Sharing System: Foreword,” The Bell System Technical Journal, volume 57, number 6, pages 1899–1904, July 1978.
Eric S. Raymond: The Art of UNIX Programming. Addison-Wesley, 2003. ISBN: 978-0-13-142901-7
Ronald Duncan: “Text File Formats – ASCII Delimited Text – Not CSV or TAB Delimited Text,” ronaldduncan.wordpress.com, October 31, 2009.
Alan Kay: “Is 'Software Engineering' an Oxymoron?,” tinlizzie.org.
Martin Fowler: “InversionOfControl,” martinfowler.com, June 26, 2005.
Daniel J. Bernstein: “Two File Descriptors for Sockets,” cr.yp.to.
Rob Pike and Dennis M. Ritchie: “The Styx Architecture for Distributed Systems,” Bell Labs Technical Journal, volume 4, number 2, pages 146–152, April 1999.
Sanjay Ghemawat, Howard Gobioff, and Shun-Tak Leung: “The Google File System,” at 19th ACM Symposium on Operating Systems Principles (SOSP), October 2003. doi:10.1145/945445.945450
Michael Ovsiannikov, Silvius Rus, Damian Reeves, et al.: “The Quantcast File System,” Proceedings of the VLDB Endowment, volume 6, number 11, pages 1092–1101, August 2013. doi:10.14778/2536222.2536234
“OpenStack Swift 2.6.1 Developer Documentation,” OpenStack Foundation, docs.openstack.org, March 2016.
Zhe Zhang, Andrew Wang, Kai Zheng, et al.: “Introduction to HDFS Erasure Coding in Apache Hadoop,” blog.cloudera.com, September 23, 2015.
Peter Cnudde: “Hadoop Turns 10,” yahoohadoop.tumblr.com, February 5, 2016.
Eric Baldeschwieler: “Thinking About the HDFS vs. Other Storage Technologies,” hortonworks.com, July 25, 2012.
Brendan Gregg: “Manta: Unix Meets Map Reduce,” dtrace.org, June 25, 2013.
Tom White: Hadoop: The Definitive Guide, 4th edition. O'Reilly Media, 2015. ISBN: 978-1-491-90163-2
Jim N. Gray: “Distributed Computing Economics,” Microsoft Research Tech Report MSR-TR-2003-24, March 2003.
Márton Trencséni: “Luigi vs Airflow vs Pinball,” bytepawn.com, February 6, 2016.
Roshan Sumbaly, Jay Kreps, and Sam Shah: “The 'Big Data' Ecosystem at LinkedIn,” at ACM International Conference on Management of Data (SIGMOD), July 2013. doi:10.1145/2463676.2463707
Alan F. Gates, Olga Natkovich, Shubham Chopra, et al.: “Building a High-Level Dataflow System on Top of Map-Reduce: The Pig Experience,” at 35th International Conference on Very Large Data Bases (VLDB), August 2009.
Ashish Thusoo, Joydeep Sen Sarma, Namit Jain, et al.: “Hive – A Petabyte Scale Data Warehouse Using Hadoop,” at 26th IEEE International Conference on Data Engineering (ICDE), March 2010. doi:10.1109/ICDE.2010.5447738
“Cascading 3.0 User Guide,” Concurrent, Inc., docs.cascading.org, January 2016.
“Apache Crunch User Guide,” Apache Software Foundation, crunch.apache.org.
Craig Chambers, Ashish Raniwala, Frances Perry, et al.: “FlumeJava: Easy, Efficient Data-Parallel Pipelines,” at 31st ACM SIGPLAN Conference on Programming Language Design and Implementation (PLDI), June 2010. doi:10.1145/1806596.1806638
Jay Kreps: “Why Local State is a Fundamental Primitive in Stream Processing,” oreilly.com, July 31, 2014.
Martin Kleppmann: “Rethinking Caching in Web Apps,” martin.kleppmann.com, October 1, 2012.
Mark Grover, Ted Malaska, Jonathan Seidman, and Gwen Shapira: Hadoop Application Architectures. O'Reilly Media, 2015. ISBN: 978-1-491-90004-8
Philippe Ajoux, Nathan Bronson, Sanjeev Kumar, et al.: “Challenges to Adopting Stronger Consistency at Scale,” at 15th USENIX Workshop on Hot Topics in Operating Systems (HotOS), May 2015.
Sriranjan Manjunath: “Skewed Join,” wiki.apache.org, 2009.
David J. DeWitt, Jeffrey F. Naughton, Donovan A.Schneider, and S. Seshadri: “Practical Skew Handling in Parallel Joins,” at 18th International Conference on Very Large Data Bases (VLDB), August 1992.
Marcel Kornacker, Alexander Behm, Victor Bittorf, et al.: “Impala: A Modern, Open-Source SQL Engine for Hadoop,” at 7th Biennial Conference on Innovative Data Systems Research (CIDR), January 2015.
Matthieu Monsch: “Open-Sourcing PalDB, a Lightweight Companion for Storing Side Data,” engineering.linkedin.com, October 26, 2015.
Daniel Peng and Frank Dabek: “Large-Scale Incremental Processing Using Distributed Transactions and Notifications,” at 9th USENIX conference on Operating Systems Design and Implementation (OSDI), October 2010.
“"Cloudera Search User Guide," Cloudera, Inc., September 2015.
Lili Wu, Sam Shah, Sean Choi, et al.: “The Browsemaps: Collaborative Filtering at LinkedIn,” at 6th Workshop on Recommender Systems and the Social Web (RSWeb), October 2014.
Roshan Sumbaly, Jay Kreps, Lei Gao, et al.: “Serving Large-Scale Batch Computed Data with Project Voldemort,” at 10th USENIX Conference on File and Storage Technologies (FAST), February 2012.
Varun Sharma: “Open-Sourcing Terrapin: A Serving System for Batch Generated Data,” engineering.pinterest.com, September 14, 2015.
Nathan Marz: “ElephantDB,” slideshare.net, May 30, 2011.
Jean-Daniel (JD) Cryans: “How-to: Use HBase Bulk Loading, and Why,” blog.cloudera.com, September 27, 2013.
Nathan Marz: “How to Beat the CAP Theorem,” nathanmarz.com, October 13, 2011.
Molly Bartlett Dishman and Martin Fowler: “Agile Architecture,” at O'Reilly Software Architecture Conference, March 2015.
David J. DeWitt and Jim N. Gray: “Parallel Database Systems: The Future of High Performance Database Systems,” Communications of the ACM, volume 35, number 6, pages 85–98, June 1992. doi:10.1145/129888.129894
Jay Kreps: “But the multi-tenancy thing is actually really really hard,” tweetstorm, twitter.com, October 31, 2014.
Jeffrey Cohen, Brian Dolan, Mark Dunlap, et al.: “MAD Skills: New Analysis Practices for Big Data,” Proceedings of the VLDB Endowment, volume 2, number 2, pages 1481–1492, August 2009. doi:10.14778/1687553.1687576
Ignacio Terrizzano, Peter Schwarz, Mary Roth, and John E. Colino: “Data Wrangling: The Challenging Journey from the Wild to the Lake,” at 7th Biennial Conference on Innovative Data Systems Research (CIDR), January 2015.
Paige Roberts: “To Schema on Read or to Schema on Write, That Is the Hadoop Data Lake Question,” adaptivesystemsinc.com, July 2, 2015.
Bobby Johnson and Joseph Adler: “The Sushi Principle: Raw Data Is Better,” at Strata+Hadoop World, February 2015.
Vinod Kumar Vavilapalli, Arun C. Murthy, Chris Douglas, et al.: “Apache Hadoop YARN: Yet Another Resource Negotiator,” at 4th ACM Symposium on Cloud Computing (SoCC), October 2013. doi:10.1145/2523616.2523633
Abhishek Verma, Luis Pedrosa, Madhukar Korupolu, et al.: “Large-Scale Cluster Management at Google with Borg,” at 10th European Conference on Computer Systems (EuroSys), April 2015. doi:10.1145/2741948.2741964
Malte Schwarzkopf: “The Evolution of Cluster Scheduler Architectures,” firmament.io, March 9, 2016.
Matei Zaharia, Mosharaf Chowdhury, Tathagata Das, et al.: “Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing,” at 9th USENIX Symposium on Networked Systems Design and Implementation (NSDI), April 2012.
Holden Karau, Andy Konwinski, Patrick Wendell, and Matei Zaharia: Learning Spark. O'Reilly Media, 2015. ISBN: 978-1-449-35904-1
Bikas Saha and Hitesh Shah: “Apache Tez: Accelerating Hadoop Query Processing,” at Hadoop Summit, June 2014.
Bikas Saha, Hitesh Shah, Siddharth Seth, et al.: “Apache Tez: A Unifying Framework for Modeling and Building Data Processing Applications,” at ACM International Conference on Management of Data (SIGMOD), June 2015. doi:10.1145/2723372.2742790
Kostas Tzoumas: “Apache Flink: API, Runtime, and Project Roadmap,” slideshare.net, January 14, 2015.
Alexander Alexandrov, Rico Bergmann, Stephan Ewen, et al.: “The Stratosphere Platform for Big Data Analytics,” The VLDB Journal, volume 23, number 6, pages 939–964, May 2014. doi:10.1007/s00778-014-0357-y
Michael Isard, Mihai Budiu, Yuan Yu, et al.: “Dryad: Distributed Data-Parallel Programs from Sequential Building Blocks,” at European Conference on Computer Systems (EuroSys), March 2007. doi:10.1145/1272996.1273005
Daniel Warneke and Odej Kao: “Nephele: Efficient Parallel Data Processing in the Cloud,” at 2nd Workshop on Many-Task Computing on Grids and Supercomputers (MTAGS), November 2009. doi:10.1145/1646468.1646476
Lawrence Page, Sergey Brin, Rajeev Motwani, and Terry Winograd: "The PageRank"
Leslie G. Valiant: “A Bridging Model for Parallel Computation,” Communications of the ACM, volume 33, number 8, pages 103–111, August 1990. doi:10.1145/79173.79181
Stephan Ewen, Kostas Tzoumas, Moritz Kaufmann, and Volker Markl: “Spinning Fast Iterative Data Flows,” Proceedings of the VLDB Endowment, volume 5, number 11, pages 1268-1279, July 2012. doi:10.14778/2350229.2350245
Grzegorz Malewicz, Matthew H.Austern, Aart J. C. Bik, et al.: “Pregel: A System for Large-Scale Graph Processing,” at ACM International Conference on Management of Data (SIGMOD), June 2010. doi:10.1145/1807167.1807184
Frank McSherry, Michael Isard, and Derek G. Murray: “Scalability! But at What COST?,” at 15th USENIX Workshop on Hot Topics in Operating Systems (HotOS), May 2015.
Ionel Gog, Malte Schwarzkopf, Natacha Crooks, et al.: “Musketeer: All for One, One for All in Data Processing Systems,” at 10th European Conference on Computer Systems (EuroSys), April 2015. doi:10.1145/2741948.2741968
Aapo Kyrola, Guy Blelloch, and Carlos Guestrin: “GraphChi: Large-Scale Graph Computation on Just a PC,” at 10th USENIX Symposium on Operating Systems Design and Implementation (OSDI), October 2012.
Andrew Lenharth, Donald Nguyen, and Keshav Pingali: “Parallel Graph Analytics,” Communications of the ACM, volume 59, number 5, pages 78–87, May doi:10.1145/2901919
Fabian Hüske: “Peeking into Apache Flink's Engine Room,” flink.apache.org, March 13, 2015.
Mostafa Mokhtar: “Hive 0.14 Cost Based Optimizer (CBO) Technical Overview,” hortonworks.com, March 2, 2015.
Michael Armbrust, Reynold S Xin, Cheng Lian, et al.: “Spark SQL: Relational Data Processing in Spark,” at ACM International Conference on Management of Data (SIGMOD), June 2015. doi:10.1145/2723372.2742797
Daniel Blazevski: “Planting Quadtrees for Apache Flink,” insightdataengineering.com, March 25, 2016.
Tom White: “Genome Analysis Toolkit: Now Using Apache Spark for Data Processing,” blog.cloudera.com, April 6, 2016.

上一章	目录	下一章
第三部分：派生数据	设计数据密集型应用	第十章：流处理

有些人认为cat这里并没有必要，因为输入文件可以直接作为awk的参数。但这种写法让线性管道更为显眼。 ↩︎
统一接口的另一个例子是URL和HTTP，这是Web的基石。一个URL标识一个网站上的一个特定的东西（资源），你可以链接到任何其他网站的任何网址。具有网络浏览器的用户因此可以通过跟随链接在网站之间无缝跳转，即使服务器可能由完全不相关的组织维护。这个原则现在似乎非常明显，但它却是网络取能取得今天成就的关键。之前的系统并不是那么统一：例如，在公告板系统（BBS）时代，每个系统都有自己的电话号码和波特率配置。从一个BBS到另一个BBS的引用必须以电话号码和调制解调器设置的形式；用户将不得不挂断，拨打其他BBS，然后手动找到他们正在寻找的信息。这是不可能的直接链接到另一个BBS内的一些内容。 ↩︎
**巴尔干化（Balkanization）**是一个常带有贬义的地缘政治学术语，其定义为：一个国家或政区分裂成多个互相敌对的国家或政区的过程。 ↩︎
除了使用一个单独的工具，如netcat或curl。 Unix开始试图将所有东西都表示为文件，但是BSD套接字API偏离了这个惯例【17】。研究用操作系统Plan 9和Inferno在使用文件方面更加一致：它们将TCP连接表示为/net/tcp中的文件【18】。 ↩︎
一个不同之处在于，对于HDFS，可以将计算任务安排在存储特定文件副本的计算机上运行，而对象存储通常将存储和计算分开。如果网络带宽是一个瓶颈，从本地磁盘读取有性能优势。但是请注意，如果使用纠删码，则会丢失局部性，因为来自多台机器的数据必须进行合并以重建原始文件【20】。 ↩︎
我们在本书中讨论的连接通常是等值连接，即最常见的连接类型，其中记录与其他记录在特定字段（例如ID）中具有相同值相关联。有些数据库支持更通用的连接类型，例如使用小于运算符而不是等号运算符，但是我们没有地方来讲这些东西。 ↩︎
这个例子假定散列表中的每个键只有一个条目，这对用户数据库（用户ID唯一标识一个用户）可能是正确的。通常，哈希表可能需要包含具有相同键的多个条目，并且连接运算符将输出关键字的所有匹配。 ↩︎

115 KiB Raw Blame History Unescape Escape