在前一篇文章中（链接参加文末），我们介绍了map端Join操作的几大方法。一般情况下，我会推荐企业选择map端的Join操作，这可以节省不小的成本。但是，如果数据集过于庞大以至于没有合适的map端连接方法适用，则需要使用MapReduce中的shuffle对数据进行排序和连接，并考虑选择Reduce端的Join操作。

一、重分区Join操作（Reduce端）

本文介绍的第一种方法是最基本的重分区Join操作，该方法允许执行内部和外部Join。开始之前，我们先搞清楚要解决的问题是将大型数据集Join在一起，我们选用的解决方案是Reduce端重分区Join。该方法是一种Reduce端Join实现，利用MapReduce的sortmerge将记录组合在一起，作为单个MapReduce作业实现，可支持N路连接，其中N是要连接的数据集数量。

Map端负责从数据集中读取数据，确定每个Join操作的value，并将该value的key输出，输出key包含在reducer中并将数据集组合在一起以生成最终结果。

单个reducer调用接收map函数Join操作发出的Key对应的所有值，并将数据分N个分区，其中N是要连接的数据集数量。reducer读取连接value的所有输入并将它们分区到内存中，然后跨所有分区执行笛卡尔积，并发出每个Join操作的结果。

图6.10 重分区Join操作的基本MapReduce实现

MapReduce代码要支持这种技术，需要满足以下条件：

支持多个map类，每个map类处理不同的输入数据集，这是通过使用MultipleInputs类完成的。需要一种方法来标记mapper发出的记录，以便可以与其原点的数据集相关联，本文将使用htuple项目处理MapReduce中的数据。

重分区Join操作的代码如下：

可以使用以下命令运行作业并查看输出：

总结

Hadoop捆绑了一个hadoop-datajoin模块，这是一个重分区Join操作框架，包括用于处理多个输入数据集和执行Join操作的管道。上述操作示例及hadoop-datajoin代码是重分区Join的最基本形式，两者都要求在执行笛卡尔积之前将连接key的所有数据加载到内存中，但如果连接key的基数大于可用内存，那么，这种方法就不太适用。下一个技术将着眼解决此问题。

二、优化重分区Join操作

旧版重分区Join操作实现会浪费大量空间，需要将给定key的所有value加载到内存中才能执行多路连接，将较小的数据集加载到内存中才能迭代更大的数据集，沿途执行Join更有效。

我们希望在MapReduce中执行重分区Join，且无需缓存reducer中的所有记录。优化后的重分区Join框架将仅缓存要连接的其中一个数据集，以减少reducer中缓存的数据量。此优化仅缓存来自两个数据集中较小者的记录，以减少缓存所有记录的内存开销，图6.11显示了改进的重分区Join实现。

图6.11 重分区Join操作优化MapReduce实现

该技术与旧版相比存在一定差异，此处使用辅助排序确保来自较小数据集的所有记录在较大数据集的记录之前到达reducer，以此来尽可能减少reducer中要缓存的数据量。此外，mapper会发出需要进行Join操作的用户名元组的key以及标识原始数据集的字段。

以下代码显示了一个新的枚举，显示了用户mapper如何填充元组字段：

需要更新MapReduce驱动程序代码以指示元组中的哪些字段应用于排序、分区和分组：

分区程序应仅基于用户名进行分区，以便用户的所有记录都到达同一个reducer。排序应使用用户名和数据集指示符，以便首先排序较小的数据集(由于USERS常量小于USER_LOGS常量，导致用户记录在用户登录之前排序)。分组应对用户进行分组，以便将两个数据集都流式传输到同一个reducer调用：

最后，我们要修改reducer以缓存传入的用户记录，然后将其与用户日志Join：

可以使用以下命令来运行作业并查看输出：

Hive

在执行重分区Join操作时，Hive可支持类似优化。Hive可缓存Join键的所有数据集，然后流式传输大型数据集，使其不需要存储在内存中。假定在查询时，Hive最后指定的数据集最大。想象一下，你有两个名为users和user_logs的表，而user_logs要大得多。要连接这些表，我们需要确保user_logs表被引用为查询中的最后一个：

如果不想重新查询，可以使用STREAMTABLE提示告诉Hive哪个表更大：

总结

此操作实现通过仅缓冲较小数据集的value来改进早期技术，但它仍然存在数据在map和reducer之间的传输问题，这是一个昂贵的网络成本。此外，旧版可以支持N路连接，但是这种实现仅支持双向连接。

三、使用Bloom过滤器来减少混洗数据

如果希望根据某些谓词对数据子集执行Join操作，例如“仅限居住在加利福尼亚地区的用户”。到目前为止，我们还必须在reducer中执行过滤器才可以实现这一目的 ，因为只有一个数据集存放了有关状态的详细信息——用户日志没有该信息。接下来，我将介绍如何在map端使用Bloom过滤器，这会对作业执行时间产生很大影响。我要解决的问题是在重分区Join操作中过滤数据，但要将该过滤器推送到mapper。一个可行的解决方案是使用预处理作业创建Bloom过滤器，然后在重分区作业中加载Bloom过滤器以过滤mapper中的记录。

Bloom过滤器是一种非常有用的随机数据结构，它利用位数组简洁表明集合，并能判断一个元素是否属于该集合。然而，与Java中的HashSet相比，Bloom需要的内存要少得多，因此它们非常适合处理大型数据集。此解决方案有两个步骤，一是运行作业来生成Bloom过滤器，该过滤器将对用户数据进行操作，并由居住在加利福尼亚地区的用户填充；二是在重分区Join操作中使用此Bloom过滤器丢弃不需要的用户，该过程需要Bloom过滤器的原因是用户日志的mapper没有状态的详细信息。

图6.12 在重分区Join中使用Bloom过滤器的两步过程

第1步：创建Bloom过滤器

第一个作业是创建Bloom过滤器，其中包含加利福尼亚州的用户名。mapper生成中间Bloom过滤器，reducer将其组合成一个Bloom过滤器，作业输出是包含序列化Bloom过滤器的Avro文件：

第2步：重分区Join

重分区Join与上文提到的唯一区别是mapper加载第一步中生成的Bloom过滤器，并且在处理map记录时，执行针对Bloom过滤器的元素审查以确定是否应将记录发送给reducer。以下代码显示了两件事：一般化Bloom过滤器加载、抽象mapper以及支持两个Join数据集的子类：

以下命令运行两个作业并转储Join输出：

总结

该技术提出了一种在两个数据集上执行map端过滤的有效方法，以最小化mapper和reducer之间的网络I/O。作为shuffle的一部分，它还减少了mapper和reducer的磁盘溢出数据量。过滤器通常是加速和优化作业最简单有效的方法，重分区Join也同样适用于其他MapReduce作业。

四、reducer端Join操作可能发生数据倾斜

数据倾斜是实际操作中很容易碰到的问题，可能存在两种类型的数据倾斜：

高Join-key基数，其中有一些连接key在一个或两个数据集中具有大量记录，我把这种称之为join-product偏差。糟糕的散列分区，少数reducer在总记录数中占很大比例，我将此称为散列分区倾斜。

五、加入具有高连接密钥基数的大型数据集

这种技术解决了join-product的倾斜问题，下一个技术检查了散列分区偏差。现在面临的问题是某些连接key是高基数的，这会导致某些reducer在尝试缓存这些key时耗尽内存。我们可以过滤掉这些key并将它们单独连接或将其溢出到reducer中并安排后续作业Join。

如果提前知道了哪些Key是高基数的，则可以将其分成单独的Join作业，如果不确定高基数Key是哪些，则可能需要在reducer中构建智能检测并将其写入副本文件，该文件由后续作业Join，如图6.14所示。

图6.13 提前知道高基数密钥时处理倾斜

图6.14 提前知道高基数密钥处理时的偏差

Hive

Hive支持类似于第二种方法的偏斜缓解策略，运行作业之前可指定以下配置启用：

可以选择设置一些其他配置来控制在高基数key上运行的map端连接：

最后，如果在SQL中使用GROUP BY，可能还需要考虑启用以下配置来处理分组数据中的偏差：

总结

此技术假设给定的Join键，只有一个数据集具有高基数出现，因此可缓存较小数据集的map端连接。如果两个数据集都是高基数的，那么将面临一个昂贵的笛卡尔积运算，执行起来会很慢，因为它不适合MapReduce的工作方式(这意味着它本身不可拆分和可并行化)。在这种情况下，我们应该重新检查是否有任何技术(如过滤或投影)可帮助减少执行join所需的时间。

六、处理由散列分区生成的偏差

MapReduce的默认分区程序是一个散列分区程序，接受每个map输出key的散列，并对reducer数量建模，以确定key被发送到哪个reducer。散列分区程序可以很好地用作通用分区程序，但是有些数据集可能会导致散列分区程序因一些不成比例的密钥散列到同一个reducer而使其重载。与大多数reducer相比，这些reducer需要更长时间才能完成。此外，当检查straggler reducer计数器时，会注意到发送给落后者的组数远远高于已完成的其他组。

区分高基数key与散列分区引起的偏差可以使用MapReduce reducer来识别数据倾斜类型。由性能较差的哈希分区器引入的偏差将具有更多的组(唯一密钥)发送到这些reducer，而导致倾斜的高基数密钥可以通过所有reducer中大致相等数量的组来证明，倾斜越多，reducer的记录数量越多。

我们要解决的问题是reducer端连接需要很长时间才能完成，而落后的组需要比大多数reducer更长时间。使用范围分区程序或编写自定义分区程序，将偏移的key集中到一组reducer。此解决方案的目标是省去默认的散列分区程序，并将其替换为可以更好处理数据倾斜的内容，本文提供两个选项可供探索：

使用与Hadoop捆绑在一起的sampler和TotalOrderPartitioner，将散列分区程序替换为范围分区程序。编写自定义分区程序，将具有数据倾斜的key路由到为倾斜key保留的Reducer。

范围分区法

范围分区根据预定义值分配map输出，其中每个map接收该范围内的所有reducer，这正是TotalOrderPartitioner的工作原理。实际上，TeraSort使用TotalOrderPartitioner在所有Reducer之间均匀分布，以最大限度减少数据倾斜。TotalOrderPartitioner附带采样器，可对输入数据进行采样并将其写入HDFS，然后在分区时由TotalOrderPartitioner使用。

自定义分区法

如果已经知道哪些Key显示数据倾斜，并且该组Key是静态的，则可以编写自定义分区程序以将这些高基数key推送到一组reducer。

总结

在上述两种方法中，范围分区可能是最佳解决方案，因为大多数情况下，我们可能不知道哪些Key是倾斜的，并且表现出倾斜的key也可能随时间而变化。MapReduce中可能有reducer端连接，因为它们将map输出Key排序并关联在一起。在之后的文章中，我们将介绍MapReduce相关的排序技术。