Interpreting the Data:Parallel Analysis with Sawzall（1）

作者
Rob Pike, Sean Dorward, Robert Griesemer, Sean Quinlan

Google, Inc.

(Draft submitted to Scientific Programming Journal)



概要
超大量的数据往往会采用一种平面的正则结构，存放于跨越多个计算机的多个磁盘上。这方面的例子包括了电话通话记录，网络日志，web文档库等等。只要这些超大量的数据集不能装在单个关系数据库里边的时候，传统的数据库技术对于研究这些超大数据集来说那就是没有意义的。此外，对于这些数据集的分析可以展示成为应用简单的，便于分布式处理的计算方法：比如过滤，聚合，统计抽取，等等。我们在这里介绍这样一种这样的自动化分析系统。在过滤阶段，查询请求通过一种全新的编程语言来快速执行，把数据处理到聚合阶段。无论过滤阶段还是聚合阶段都是分布在上百台甚至上千台计算机上执行的。他们的结果通过比较并且保存到一个文件。这个系统的设计-包括分成两阶段，以及这种新式的编程语言，聚合器的特性-都是在数据和计算分布在很多台机器上的情况下，内嵌使用并行机制的。



1．介绍
有不少数据集都是超大的，或者非常动态，或者就是因为太笨拙了，而不能有效地通过关系数据库进行管理。典型的场景是一组大量无格式文件-有时候是上petabytes(2的50次方1,125,899,906,842,624)-分布在多个计算机上的多个磁盘上。这些文件都包含了无数的记录，这些记录是通常会通过一个轴来组织，比如通过时间轴或者地理轴进行组织。例如：这堆文件可能包含一个web网页仓库，用来构造internet搜索引擎的索引系统，或者这堆文件用来记录上千台在线服务器的健康日志，或者用来记录电话呼叫记录或者商业交易日至，网络包记录，web服务器查询记录，或者高级一点的数据比如卫星图像等等。但是对这些数据的分析经常可以表示成为简单的操作，远比普通SQL查询要简单得操作来完成。举一个例子，我们通常会统计满足某条件的记录数，或者抽取这些记录，或者查询异常记录，或者构造记录中某一个域值的频率柱状图。另一方面，查询也可能会较为复杂，但是这些查询依旧可以展示成为通过一系列简单查询来完成，这些简单查询都可以简单映射到这些文件的记录集上。




图1：5组机架，每组有50-55台计算机，每台计算机有4个磁盘。这样一个架构可以有到250TB的待分析数据量。我们可以在250台以上的计算机上分别执行过滤来极大的的提高并行度，并且把他们的结果通过网络汇聚到一起（参见弧线）



由于数据记录存放在多台计算机上，那么用这些计算机本身的能力来进行分析的方法就相当有效。特别是，当单独每一个步骤都可以表示成为每次对独立的记录进行操作的时候，我们就可以把计算分布到所有这些机器上，这样就能达到相当高的吞吐量。（前边提及的每个例子都有这样的特点）。这些简单操作都要求一个聚合的阶段。例如，如果我们统计记录数，我们需要把每一个机器统计出来的记录数相加，作为最终的输出结果。

所以，我们把我们的计算分成两个阶段。第一个阶段我们对每一条记录分别计算，第二个阶段我们聚合这些结果（图2）。本论文描述的系统更进一步考虑了这个问题。我们用一个全新的编程语言来进行第一个阶段的分析，从处理粒度上，它一次处理一条记录，并且在阶段2严格限制预先定义的处理阶段1产出物的聚合器处理的集合。通过约束本模式的计算量，我们可以达到非常高的吞吐量。虽然并非所有的计算都能适合这样的模式，但是仅仅通过不多的代码就能够驱动上千台机器并行计算还是很划算的。




              RAW DATA

图2：总体数据流图，过滤，聚合和比较。每一步都比上一步产生更少的数据。



当然，我们还有很多小问题要解决。计算必须要分解成为小块并且分布到每一个存储数据的节点上进行执行，尽量让计算和数据在一台机器上以避免网络瓶颈。由于使用的机器越多，那么越有可能有机器会在运算中宕机，所以，必须系统必须要有容错能力。这些都是困难但是有趣的问题，但是他们都必须能够在没有人为干预的情况下完成。Google有好几个这样的基础架构，包括GFS[9]和MapReduce[8]，通过容错技术和可靠性设计来提供了一个非常强大的框架，可以用来实现一个很大的，分布式处理的并行系统。因此我们着重于我们的目标：清晰的表达分析处理，并且迅速执行分析处理。



2．总览
简要而言，我们的系统通过处理用户提交的用特别设计的编程语言写成的查询，并发的在分布到大量机器上的记录集中，进行记录级别的查询，并且搜集查询结果，通过一组高性能的聚合器进行查询结果的汇聚。这两部发呢别执行，通常分布到不同的计算机集群上。

这样的处理典型类型是并发处理分布在成百上千台计算机上的gigabyte或者数Tbyte数据。一个简单的分析可能需要花去一个CPU好几个月的时间，但是通过上千台计算机的并行处理，只需要几个小时的时间就能处理完。

有两个条件决定着系统的设计。首先，如果查询操作是对记录间可交换的，就是说记录处理的先后顺序是不重要的。我们于是可以用任意的顺序来处理这个查询操作。第二，如果聚合操作是可交换的，中间结果的处理顺序是不重要的。此外，如果他们也是可结合的，中间处理结果可以被任意分组或者分成不同的步骤进行聚合。举一个例子，对于统计数量包括汇总数量来说，无论中间结果如何的累加或者分组结合累加，他们最终的结果都不会受到影响。这个交换性和结合性的约束并不算过分苛刻，他们可以提供很广阔的查寻范围，包括：统计，筛选，取样，柱状图，寻找常见项目，等等。

虽然聚合器组是有限的，但是对于查询阶段来说，应当包括更加通用的内容，我们介绍一种新的解释执行的程序语言Sawzall[1]（解释语言的性能已经足够了：因为程序多数都是比较小的，而且他们需要处理的数据往往很大，所以往往是受I/O的限制，这在性能的章节有所讨论）

一个分析操作如下：首先输入被分解成为要被处理的数据小块，也许是一组独立的文件或者一组记录，这些记录或者文件分布于多个存储节点上。数据小块可以远远多于计算机的数量。

其次，Sawzall解释器开始处理每一个小块数据。这个处理跨越了大量机器，也许数据和机器绑定在一起，也可能数据在临近的机器上而不在一起。

Sawzall程序分别处理每一个输入记录。每一个记录的输出结果，0个或者多个中间结果值-整数，字串，key-value pairs，tuple等等-将和其他记录的输出值合并。

这些中间结果于是被发送到运行聚合器的进一步处理的结点上，这些节点比较和减少中间结果，并且构造终结结果。在一个典型的运行中，主要的计算机集群会运行Sawzall，并且小一点的集群会运行聚合器，这样的结构反映不仅是体现在计算量的差异，也体现在网络负载的均衡考虑；每一个步骤，数据流量都比上一个步骤要少（参见图2）。

当所有的处理都完成之后，结果将被排序，格式化，并且保存到一个文件。



3．例子
用这个简单的例子可以更清楚的表达这样的想法。我们说我们的输入是一个由浮点数记录组成的文件集合。这个完整的Sawzall程序将会读取输入并且产生三个结果：记录数，值得总合，并且值得平方和。

count:            table sum of int;

total:                     table sum of float;

sum_of_squares:   table sum of float;

x:                  float=input;

emit        count<-1;

emit        sum<-x;

emit        sum_of_squares <- x*x;



前三行定义了聚合器：计数器，合计，平方和。关键字table定义了聚合器类型；在Sawzall中，即使聚合器可能是单例的，也叫做table。这些特定的table是属于合计的table；他们把输入的整数或者浮点数的值进行累加。

对于每一个输入的记录，Sawzall初始化预定义的变量input来装载尚未处理的输入记录二进制串。因此，行：

x: float = input;

把输入记录从它对外的表示转换成为内嵌的浮点数，并且保存在本地变量x。最后，三个emit语句发送中间结果到聚合器。

当程序执行的时候，程序对每一个输入记录只执行1次。程序定义的本地变量每次重新创建，但是程序定义的table会在所有执行中共享。处理过的值会通过全局表进行累加。当所有记录都处理了以后，表中的值保存在一个或者多个文件中。

接下来的几节讲述了本系统所基于的部分Google的基础架构：协议buffers，Google文件系统，工作队列，MapReduce。后续章节描述语言和其他系统的详尽部分。











4．协议Buffer
虽然在最开始已经考虑了定义服务器之间的消息通讯，Google的协议Buffer也同样用于描述保存在磁盘的持久化存储的记录格式。

这个协议Buffer的用处很类似XML，但是要紧凑的多，通过一个二进制的表示以及一个外部的数据描述语言（DataDescription Language DDL）是的协议编译器能够把协议编译成为各种语言的支持代码。

DDL构造一个清晰，紧凑，可扩展的针对二进制记录的描述，并且对记录的字段进行命名。虽然二进制格式已经是相当紧凑的，但是常常还会在保存到磁盘的时候再进行一个压缩，包裹一个压缩层。

协议编译器读取DDL描述并且产生用于对数据的：组织，访问，列集及散列处理的代码。编译时候的标志指定了输出的语言：C++，Java，Python,等等。这个产生的代码通过嵌入应用程序中，能够提供对数据记录高效简洁的访问。同时，也应该提供验证和调试保存的协议buffer的工具包

我们系统操作的大部分数据集都是按照协议buffer约定的格式存储的记录。协议编译器通过增加对Sawzall的扩展来提供在新语言下的协议buffer的高效IO性能。



5．Google文件系统（GFS）
我们系统访问的数据集通常保存在GFS内，就是Goole的文件系统[9]。GFS提供一个可靠的分布式存储系统，它可以通过分布在上千台计算机的64M”块”组织成为上Petabyte级别的文件系统。每一个块都有备份，通常是3个备份，在不同的计算机节点上，这样GFS可以无缝的从磁盘或者计算机的故障中容错。

GFS是一个应用级别的文件系统，有着传统的分级的命名机制。数据集本身通常用一个常规的结构存放，这些结构存放在很多独立的GFS文件中，每一个GFS文件大小接近1G。例如，一个文档仓库（web搜索器机器人探索结果），包含数十亿HTMLpages，可能会存放在上千个文件中，每一个文件压缩存放百万级别的文档，每个文档大概有数K字节大小。



6．工作队列和MapReduce
把工作安排到一组计算机集群上进行工作的处理软件叫做（稍稍有点容易误解）工作队列。工作队列很有效的在一组计算机及其磁盘组上创建了一个大尺度的分时共享机制。它调度任务，分配资源，报告状态，并且汇集结果。

工作队列和Condor[15]等其他系统比较类似。我们经常把工作队列集群和GFS集群部署在相同的计算机集群上。这是因为GFS是一个存储系统，CPU通常负载不太高，在CPU空闲阶段可以用来运行工作队列任务。

MapReduce[8]是一个运行在工作队列上的应用程序库。它提供三个主要功能。首先，它提供一个给予大量数据的并行处理的程序运行模式。第二，它把应用从运行在分布式程序的细节中隔离出来，包括类似数据分布，调度，容错等等。最后，当发现条件许可时，各个计算机或者存储自己的GFS数据节点上的应用程序可以执行计算，减少网络的流量。

就是MapReduce名字说明的含义，这个执行模式分成两个步骤：第一个步骤是把一个执行操作映射到数据集合的全部元素；第二个步骤是简化第一个步骤地输出结果，并且产生最终的应答。例如，一个使用MapReduce的排序程序将会映射一个标准的排序算法到数据集和的每一个文件上，接下来就是运行一个合并排序程序来简化第一个步骤出来的单独结果，并且产生最终地输出。在上千台机器的Cluster中，一个MapReduce程序可以用每秒排序1G数据的速度排序上TB的数据[9]。



我们的数据处理系统是基于MapReduce的最上层的。Sawzall解释器运行在映射步骤。这是在大量机器上并发完成的，每一个执行实例处理一个文件或者一个GFS块。Sawzall程序对每一个数据集的记录执行只执行一次。映射步骤地输出是一个数据项的集合，并且是交给聚合器去处理。聚合器在简化/减少的步骤运行来合并结果成为最终的输出结果。

接下来的章节讲述这些处理的细节。



本文来自CSDN博客，转载请标明出处：http://blog.csdn.net/zhujt1981/archive/2008/04/04/2249911.aspx