Data-Intensive Text Processing with MapReduce第三章（6）-MapReduce算法设计-3.5相关连接（RELATIONAL JOINS）

3.5相关连接（RELATIONAL JOINS）

Hadoop的一个流行应用领域是数据仓库。在一个企业级的环境中，一个数据仓库作为大量数据的存储地点，存储着从销售交易到商品清单几乎所有的信息。一般来说这些数据都是相关的，但是随着数据的日益增长，数据仓库被用来像存储无结构数据那样存储半结构化的数据（例如，查询日志）。数据仓库组成了提供决策支持的商业智能应用程序的基础。普遍认为知识是通过对历史、现在的数据进行挖掘得到的，预测后的数据能带来市场上的竞争优势。

一般来说，数据仓库已经通过关系数据库实现了，特别是经过优化后的联机分析处理(OLAP)。多维度出现在平行数据库中，但用户们发现他们不能通过一个系统今天要处理的数据来衡量成本效益。并行数据库往往是很贵的---好几万一TB的用户数据。经过这几年的发展，Hadoop作为一个数据仓库被广受欢迎。Hammerbacher[68],讨论过Facebook在Oracle数据库上搭建过商业智能应用，后来放弃了，因为喜欢使用自家开发的基于Hadoop的Hive（现在是个开源项目）。Pig [114]是用Hadoop建立的进行大量数据分析的平台并且可以像处理半结构化数据那样处理结构化数据。它原先是由雅虎开发的，但现在是个开源项目。

如果想在样的环境下用Hadoop建立成功的数据仓库和复杂的分析查询应用程序，查看MapReduce怎样控制相关数据是有意义的。这节主要关注于如何在MapReduce中实现相关连接。我们要在这里强调虽然Hadoop已经被应用于处理相关数据，但Hadoop不是一个数据库。并行数据库和MapReduce在OLAP应用的环境下那个更有优势的争论还在继续。Dewitt和Stonebraker，两个在数据库社区的知名人物，在博客中提出MapReduce是巨大的退步11。作为同事，他们做了一系列论证面向行的并行数据库优于Hadoop的基准测试[120, 144]。然而，看看Dean和Ghemawat的反论[47]和不久前尝试的混合架构[1]。

我们必须停止讨论这件充满活力的事，取代它的是关注于讨论算法。从一个应用程序的观点来看，很有可能一个基于数据仓库的数据分析根本不用写MapReduce程序（实际上，基于Hadoop的Hive和Pig都用一种更高级的语言来处理大规模的数据）。然而，了解构成它的基层算法也是有益的。

这个章节展示三个不同的两个数据集之间的相关连接的策略，一般命名为S和T。让我们假设关系S像下面那样：

(k1, s1,S1)

(k2, s2, S2)

(k3, s3, S3)

…

k是我们希望连接的键，sn是数组的唯一id，在sn后的Sn表示数组中的其它属性（在连接中不重要）。同样，假定关系T像下面那样：

(k1, t1,T1)

(k3, t2,T2)

(k8, t3,T3)

…

k是连接键，tn是数组唯一的id，在tn后的Tn表示元组中的其它属性。

为了使这个任务更加具体，我们举一个现实中可能出现的情况：S代表用户资料的集合，k在这个例子中作为主键（例如，用户的id号）。这个元组可能包含人口统计资料例如年龄、性别、收入等。另一个数据集，T，代表用户网上操作的日志。每一个元组相当于用户在一个页面浏览中的网页URL和包含的额外信息，例如在网页上花的时间，产生的广告收入等。k在这个元组中可以认为是用户浏览不同网页的数据集的外键。连接这两个数据集就可以进行分析，例如，根据用户特性来终止他的网络活动（比如一些恶意攻击）。

3.5.1 REDUCE端连接

第一个相关连接的方法是在reduce里面进行连接。这个想法非常简单：我们遍历两个数据集然后发送连接键作为中间键，元组本身作为值。因为MapReduce保证同一键的所有值都会集中起来，所有元组通过连接键分组---这在我们的连接操作中不是必须的。这个方法在数据库社区中称为并行排序合并连接（parallel sort-merge join），详细说来，还有三种情况需要考虑。

首先是最简单的一对一连接，S中的最多一个元组和T中的一个元组共享同一连接键（但可能有这种情况，即S中没有和T共享一个连接键，或反过来）。在这个例子中，上面所说的算法可以正常工作。Reducer将收到像下面那样的键和值列表：

k23 → [(s64, S64), (t84,T84)]

k37 → [(s68, S68)]

k59 → [(t97,T97), (s81, S81)]

k61 → [(t99,T99)]

…

因为我们把连接键作为中间键发送出去，我们可以把它从值中去除以节省空间12。如果一个键有两个值，则其中一个一定来自S另一个一定来自T。然而，回忆在基本的MapReduce编程模式中，并不能保证值的顺序，所有第一个值可能来自S也可能来自T。我们可以继续连接两个数据集和执行额外的计算（例如，用其它属性过滤，计算总数等）。如果一个键只对应一个值，这意味着没有另外数据集的元组和它有相同的连接键，所以reducer什么都不用做。

12中间数据是否压缩并不是很重要

现在让我们考虑下一对多的连接。假设S中的元组有唯一的连接键（即，k是S中的主键），所以S是“一”T是“多”。上面说到的算法仍然可以工作，但当在reducer中处理每一个键时，我们不知道什么时候会遇到与S的元组相关的值，因为值是任意排序的。最简单的解决方法是在内存中缓存所有值，从S中挑选元组，然后与T中的每一个元组相交来执行连接。然而，我们之前已经多次遇到这种情况，它会带来伸缩性瓶颈，因为我们可能没有足够的内存来存储所有相同连接键的元组。

这个问题需要使用二次排序，解决方案就是我们刚才说到的“键值转换”模式。

在mapper中，我们创建一个包含连接键和元组id（从S或T中获得）的混合键而不是简单的把连接键作为中间键发送出去。还有两个地方需要改变：第一，我们必须定义键的顺序来让它先通过连接键排序再通过S的元组id排序最后用T的元组id排序。第二，我们必须定义partitioner来跟踪连接键，使所有有着相同连接键的混合键传到同一reducer中。

应用了“键值转换”模式后，reducer将得到类似下面那样的键值：

(k82, s105) → [(S105)]

(k82, t98) → [(T98)]

(k82, t101) → [(T101)]

(k82, t137) → [(T137)]

…

因为连接键和元组id都在中间键中，我们可以在值中移除它们来节省空间13。当reducer遇到一个新的连接键时，它能保证关联的值是从S中得到的元组。Reducer可以在内存中保存这个元组然后在下一步（直到遇到一个新的连接key）中和T中的元组交互。因为有mapreduce框架来执行排序，所以就不用再缓存元组（与S中的单个不同）。因此，我们消除了伸缩性瓶颈。最后，让我们考虑多对多的连接情况。假设S是个小数据集，上面的算法依然可以工作。想一下在reducer那里会发生什么。

13再次说明，中间数据是否压缩并不是很重要

(k82, s105) → [(S105)]

(k82, s124) → [(S124)]

…

(k82, t98) → [(T98)]

(k82, t101) → [(T101)]

(k82, t137) → [(T137)]

…

与S的所有元组连接键相同的会先遇到，reducer可以在内存中缓存。Reducer处理T中的每一个元组，并与S中的每一个元组交互。当然，我们假设S中的元组（有共同连接键的）可以放到内存中，这是这个算法的局限（和为什么我们希望控制排序顺序来让小的数据集先传进来）。

在reduce里面连接的基本思想是通过连接键重新分配两个数据集。这个方法并不是特别有效因为它需要在网络中清洗两个数据集。下面来介绍map端连接。

3.5.2 MAP端连接（MAP-SIDE JOIN）

假设我们通过连接键把两个数据集分类。我们可以同步扫描两个数据集来执行连接操作---这在数据库社区被称为合并连接。我们可以通过分割和排序两个数据库来实现平行化。例如，假设S和T都分成10个文件，用连接键以同种方法分割。进一步假设每个文件中的元组通过连接键分类。在这个例子中，我们需要简单的合并连接S的第一个文件和T中的第一个文件，S的第二个文件和T的第二个文件等。在一个MapReduce Job的map阶段中可以通过并行化来完成---这就是map端连接。实际上，我们遍历其中一个数据集（比较大的那个）和在mapper中读取另一个数据集的相关部分来执行合并连接14。这并不需要有reducer参与，除非程序员希望重新分配输出来进行更多的处理。

14这常常预示着这不是一个本地读取

map端的连接远远比reduce端的高效，因为它不需要通过网络来传输数据集。现实中能像期望那样能满足Map端的连接环境？在大多数情况下是这样的。原因是相关连接发生在一个工作流更广的语境中，它可能包含多个步骤。因此，被连接的数据集必须是之前处理的输出（无论是MapReduce jobs还是其它代码）。如果能预先知道工作流程和其相对不变（这都是对于成熟工作流的两个合理的假设），我们可以来让高效的map端连接成为可能（在MapReduce中，通过使用一个自定partitioner和控制键值对的排序顺序）。

对于特别的数据分析，reduce端的连接更加普遍，尽管效率低点，考虑到数据集有多个键，其中一个需要连接---然后无论数据是怎样组成的，map端的连接将需要将数据重新分配。作为选择，使用同一个mapper和reducer重新分配数据集经常是可能发生的。当然，这将导致通过网络传送数据的额外花费。

这是在Hadoop实现的MapReduce中使用map端连接需要记住的最后一个约束。我们假设需要连接的数据集由之前的MapReduce job产生，所以适用于键的约束在reducers的这些jobs中会被发送出去。Hadoop允许reducers发送值与正在处理值的输入键不同的键（即，输入和输出键不需要一样，甚至是不同的类型）15。然而，如果一个reducer的输出键和输入键不同，那么reducer输出的数据集就不需要在特定的partitioner中分割（因为partitioner应用与输入键而不是输出键）。因为map端连接基于对键的不断分割和排序，reducers用来生成参加下一个map端连接的的数据不能发送任何键除了它正在处理的那个。

3.5.3 基于内存的连接（MEMORY-BACKED JOIN）

除了之前提到的两个方法来连接相关数据并平衡MapReduce框架来连接有着同样连接键的元组。还有一个我们称为“基于内存的连接”的基于任意取得探索的同类型方法。最简单的版本是当两个数据集的其中一个在每一个节点中都小于内存时。在这个解决方案中，我们可以把比较小的数据集在每个mapper读取到内存中，基于连接键来获得一个关联数组使减少对元组的任意访问。Mapper的初始化API（看3.11节）可用于这个目的。Mapper然后应用到另一个（大的）数据集，对于每一个输入键值对，mapper检索在内存中的数据集看看是否有一个元组的连接键和它匹配。如果匹配的话就执行连接。这在数据库社区被认为是一个简单的哈希连接[51]。

如果内存放不下其中任意一个数据集呢？最简单的办法是把它分成更小的数据集，即把S分成n个部分,即S = S1 ∪ S2 ∪ .. . ∪Sn。我们可以通过定义n的大小来使每一个部分刚好是内存的大小，然后运行n个基于内存的连接。这样的话，当然，需要流动到另一数据集n次。

15作为对比，2.2节中讲到了Google的实现方法，reducer的输出键必须和它的输入键类型一致。

还有一个替代方案来让基于内存的连接适用与所有数据集都大于内存的情况。一个分布式的键值存储器可以通过多台机器在内存中保存一个数据集，然后去映射其它的数据集。Mapper然后并行地查询这个分布式键值存储器，如果连接键匹配则执行连接16。开源的缓存系统memcached适用于这种情况，因此我们把这种方法称为memcached连接。更多有关这个方法的信息参考技术报告[95]。