《大规模分布式存储系统》第13章 大数据【13.3】
13.3 MapReduce扩展MapReduce框架有效地解决了海量数据的离线批处理问题，在各大互联网公司得
到广泛的应用。事实已经证明了MapReduce巨大的影响力，以至于引发了一系列的扩
展和改进。这些扩展包括：
●Google Tenzing：基于MapReduce模型构建SQL执行引擎，使得数据分析人员可
以直接通过SQL语言处理大数据。
●Microsoft Dryad：将MapReduce模型从一个简单的两步工作流扩展为任何函数
集的组合，并通过一个有向无环图来表示函数之间的工作流。
●Google Pregel：用于图模型迭代计算，这种场景下Pregel的性能远远好于
MapReduce。
13.3.1 Google Tenzing
Google Tenzing是一个构建在MapReduce之上的SQL执行引擎，支持SQL查询且能
够扩展到成千上万台机器，极大地方便了数据分析人员。
1.整体架构
Tenzing系统有四个主要组件：分布式Worker池、查询服务器、客户端接口和元
数据服务器，如图13-2所示。
图 13-2 Tenzing整体架构
●查询服务器（Query Server）：作为连接客户端和worker池的中间桥梁而存在。
查询服务器会解析客户端发送的查询请求，进行SQL优化，然后将执行计划发送给分
布式Worker池执行。Tenzing支持基于规则（rule-based optimizer）以及基于开销
（cost-based optimizer）两种优化模式。
●分布式Worker池：作为执行系统，它会根据查询服务器生成的执行计划运行
MapReduce任务。为了降低查询延时，Tenzing不是每次都重新生成新进程，而是让进
程一直处于运行状态。Worker池包含master和worker两种节点，其中，master对应
MapReduce框架中的master进程，worker对应MapReduce框架中的map和reduce进程。
另外，还有一个称为master监听者（master watcher）的守护进程，查询服务器通过
master监听者获取master信息。
●元数据服务器（Metadata Server）：存储和获取表格schema、访问控制列表
（Access Control List,ACL）等全局元数据。元数据服务器使用Bigtable作为持久化的
后台存储。
●客户端接口：Tenzing提供三类客户端接口，包括API、命令行客户端（CLI）以
及Web UI。
●存储（Storage）：分布式worker池中的master和worker进程执行MapReduce任务
时需要读写存储服务。另外，查询服务器会从存储服务获取执行结果。
2.查询流程
1）用户通过Web UI、CLI或者API向查询服务器提交查询。
2）查询服务器将查询请求解析为一个中间语法树。
3）查询服务器从元数据服务器获取相应的元数据，然后创建一个更加完整的中
间格式。
4）优化器扫描该中间格式进行各种优化，生成物理查询计划。
5）优化后的物理查询计划由一个或多个MapReduce作业组成。对于每个
MapReduce作业，查询服务器通过master监听者找到一个可用的master,master将该作业
划分为多个任务。
6）空闲的worker从master拉取已就绪的任务。Reduce进程会将它们的结果写入
到一个中间存储区域中。
7）查询服务器监控这些中间存储区域，收集中间结果，并流失地返回给客户
端。
3.SQL运算符映射到MapReduce
查询服务器负责将用户的SQL操作转化为MapReduce作业，本节介绍各个SQL物
理运算符对应的MapReduce实现。
（1）选择和投影
选择运算符σ C （R）的一种MapReduce实现如下。
Map函数：对R中的每个元素t，检测它是否满足条件C。如果满足，则产生一
个“键-值”对（t,t）。也就是说，键和值都是t。
Reduce函数：Reduce的作用类似于恒等式，它仅仅将每个“键-值”对传递到输出
部分。
投影运算的处理和选择运算类似，不同的是，投影运算可能会产生多个相同的
元组，因此Reduce函数必须要剔除冗余元组。可以采用如下方式实现投影运算符
π S （R）。
Map函数：对R中的每个元组t，通过剔除属性不在S中的字段得到元组t'，输出一
个“键-值”对（t'，t'）。
Reduce函数：对任意Map任务产生的每个键t'，将存在一个或多个“键-值”对
（t'，t'），Reduce函数将（t'，）转换为（t'，t'），以保证对该键t'只产
生一个（t'，t'）对。
Tenzing执行时会做一些优化，例如选择运算符下移到存储层；如果存储层支持
列式存储，Tenzing只扫描那些查询执行必须的列。
（2）分组和聚合
假定对关系R（A,B，C）按照字段A分组，并计算每个分组中所有元组的字段B
之和。可以采用如下方式实现γ A,SUM（B） (R）。
Map函数：对于每个元组，生成“键-值”对（a,b）。
Reduce函数：每个键a代表一个分组，对与键a相关的字段B的值的列表[b 1 ，b 2 ，
…，b n ]执行SUM操作，输出结果为（a,SUM（b 1 ，b 2 ，…，b n ））。
Tenzing支持基于哈希的聚合操作，首先，放松底层MapReduce框架的限制，
shuffle时保证所有键相同的“键-值”对属于同一个Reduce任务，但是并不要求按照键
有序排列。其次，Reduce函数采用基于哈希的方法对数据分组并计算聚合结果。
（3）多表连接
大表连接是分布式数据库的难题，MapReduce模型能够有效地解决这一类问题。
常见的连接算法包括Sort Merge Join、Hash Join以及Nested Loop Join。
假设需要将R（A,B）和S（B,C）进行自然连接运算，即寻找字段B相同的元
组。可以通过Sort Merge Join实现如下：
Map函数：对于R中的每个元组（a,b），生成“键-值”对（b,（R,a）），对S中的
每个元组（b,c），生成“键-值”对（b,（S,c））。
Reduce函数：每个键值b会与一系列对相关联，这些对要么来自（R,a），要么来
自（S,c）。键b对应的输出结果是（b,[（a 1 ,b,c 1 ）,（a 2 ,b,c 2 ）,…]），也就是说，与b
相关联的元组列表由来自R和S中的具有共同b值的元组组合而成。
如果两张表格都很大，且二者的大小比较接近，Join字段也没有索引，Sort
Merge Join往往比较高效。然而，如果一张表格相比另外一张表格要大很多，Hash
Join往往更加合适。
假设R（A,B）比S（B,C）大很多，可以通过Hash Join实现自然连接。Tenzing中
一次Hash Join需要执行三个MapReduce任务。
MR1：将R（A,B）按照字段B划分为N个哈希分区，记为R 1 ，R 2 ，…，R N ；
MR2：将S（B,C）按照字段B划分为N个哈希分区，记为S 1 ，S 2 ，…，S n ；
MR3：每个哈希分区＜R i ,S i ＞对应一个Map任务，这个Map任务会将S i 加载到内
存中。对于R i 中的每个元组（a,b），生成（b,[（a,b,c 1 ）,（a,b,c 2 ）,…]），其中，
（b,）是S i 中存储的元组。Reduce的作用类似于恒等式，输出每个传入
的“键-值”对。
Sort Merge Join和Hash Join适用于两张表格都不能够存放到内存中，且连接列没
有索引的场景。如果S（B,C）在B列有索引，可以通过Remote Lookup Join实现自然
连接，如下：
Map函数：对于R中的每个元组（a,b），通过索引查询S（B,C）中所有列值为b
的元组，生成（b,[（a,b,c 1 ），（a,b,c 2 ）,…]）。
Reduce函数：Reduce的作用类似于恒等式，输出每个传入的“键-值”对。
如果S（B,C）能够存放到内存中，那么，Map进程在执行map任务的过程中会将
S（B,C）的所有元组缓存在本地，进一步优化执行效率。另外，同一个Map进程可
能执行多个map任务，这些map任务共享一份S（B,C）的所有元组缓存。
13.3.2 Microsoft Dryad
Microsoft Dryad是微软研究院创建的研究项目，主要用来提供一个分布式并行计
算平台。在Dryad平台上，每个Dryad工作流被表示为一个有向无环图。图中的每个
节点表示一个要执行的程序，节点之间的边表示数据通道中数据的传输方式，其可
能是文件、管道、共享内存、网络RPC等。Dryard工作流如图13-3所示。
图 13-3 Dryad工作流
每个节点（vertices）上都有一个处理程序在运行，并且通过数据通道
（channels）的方式在它们之间传输数据。类似于Map和Reduce函数，工作流中的
grep、sed、map、reduce、merge等函数可以被很多节点执行，每个节点会被分配一部
分输入。Dryad的主控进程（Job Manager）负责将整个工作分割成多个任务，并分发
给多个节点执行。每个节点执行完任务后通知主控进程，接着，主控进程会通知后
续节点获取前一个节点的输出结果。等到后续节点的输入数据全部准备好后，才可
以继续执行后续任务。
Dryad与MapReduce具有的共同特性就是，只有任务完成之后才会将输出传递给
接收任务。如果某个任务失败，其结果将不会传递给它在工作流中的任何后续任
务。因此，主控进程可以在其他计算节点上重启该任务，同时不用担心会将结果重
复传递给以前传过的任务。
相比多个MapReduce作业串联模型，Dryad模型的优势在于不需要将每个
MapReduce作业输出的临时结果存放在分布式文件系统中。如果先存储前一个
MapReduce作业的结果，然后再启动新的MapReduce作业，那么，这种开销很难避
免。
13.3.3 Google Pregel
Google Pregel用于图模型迭代计算，图中的每个节点对应一个任务，每个图节点
会产生输出消息给图中与它关联的后续节点，而每个节点会对从其他节点传入的输
入消息进行处理。
Pregel中将计算组织成“超步”（superstep）。在每个超步中，每个节点在上一步
收到的所有消息将被处理，并且将处理完后的结果传递给后续节点。
Pregel采用了BSP（Bulk Sychronous Parallel，整体同步并行计算）模型。每个“超
步”分为三个步骤：每个节点首先执行本地计算，接着将本地计算的结果发送给图中
相邻的节点，最后执行一次栅栏同步，等待所有节点的前两步操作结束。Pregel模型
会在每个超步做一次迭代运算，当某次迭代生成的结果没有比上一次更好，说明结
果已经收敛，可以终止迭代。
图 13-4 Pregel BSP计算模型
假设有一个带边权重的图，我们的目标是对图中的每个节点计算到其他任一节
点的最短路径长度。一开始，每个图节点a都保存了诸如（b,w）对的集合，这表示a
到b的边权重为w。
（1）超步
每个节点会将（a,b,w）传递给图中与它关联的后续节点。当节点c收到三元组
（a,b,w）时，它会重新计算c到b的最短距离，如果w+v＜u（假设当前已知的c到a的
最短距离为v,c到b的最短距离为u），那么，更新c到b的最短距离为w+v。最后，消
息（c,b,w+v）会传递给后续节点。
（2）终止条件
当所有节点在执行某个超步时都没有更新到其他节点的最短距离时，说明已经
计算出想要的结果，整个迭代过程可以结束。
Pregel通过检查点（checkpoint）的方式进行容错处理。它在每执行完一个超步之
后会记录整个计算的现场，即记录检查点情况。检查点中记录了这一轮迭代中每个
任务的全部状态信息，一旦后续某个计算节点失效，Pregel将从最近的检查点重启整
个超步。尽管上述的容错策略会重做很多并未失效的任务，但是实现简单。考虑到
服务器故障的概率不高，这种方法在大多数时候还是令人满意的。
页:
[1]