1. Map 联接优化
Index
1.1 使用分布式缓存传播哈希表文件
以前，当 2 个大数据表需要进行联接时，将有 2 个不同的 Mappers 根据联接键对这些表进行排序并发出中间文件，而 Reducer 将中间文件作为 Importing 文件并进行 true 的联接工作。此连接机制(称为“公共连接”)非常适合具有两个大数据量。但是，如果一个联接表足够小以适合 Mapper A 的内存，则无需启动 Reducer。实际上，Reducer 阶段的性能非常昂贵，因为 Map/Reduce 框架需要排序和合并中间文件。因此，Map Join 的基本思想是将小表的数据保存在 MapperA 的内存中，并在 Map 阶段进行联接工作，从而节省了 Reduce 阶段。此处显示了先前的实现(优化之前)：
图 1.以前的 Map 联接实现
在上面的图 1 中，当较大的表很大时，以前的 Map 联接实现无法很好地扩展，因为每个 Mapper 都会直接从 HDFS 读取较小的表数据。如果较大的表很大，则会启动成千上万的 Mapper 来读取较大表的不同记录。数以千计的 Mappers 会将这些小表数据从 HDFS 读取到他们的内存中，这会使对小表的访问成为性能瓶颈；或者，有时 Mappers 会因读取此小文件而超时，这可能导致任务失败。
(HIVE-1641)解决了这个问题，如下图 2 所示。
图 2.优化的 Map 联接
基本思想是在原始的“加入 Map/缩小任务”之前创建一个新任务“ MapReduce 本地任务”。这项新任务将从 HDFS 读取小表数据到内存中的哈希表。读取后，它将把内存中的哈希表序列化为磁盘上的文件，并将哈希表文件压缩为 tar 文件。在下一阶段，启动 MapReduce 任务时，它将把这个 tar 文件放到 Hadoop 分布式缓存中，它将把 tar 文件填充到每个 Mapper A 的本地磁盘上并解压缩该文件。因此，所有的 Map 器都可以将哈希表文件反序列化回内存中，并像以前一样执行连接工作。
显然，本地任务占用大量内存。因此，查询处理器将在子 jvm 中启动此任务，该子 jvm 具有与 Mapper 相同的堆大小。由于本地任务可能会用完内存，因此查询处理器将非常仔细地测量本地任务的内存使用情况。一旦本地任务的内存使用率高于阈值数。此本地任务将自身中止，并告诉用户该表太大而无法容纳在内存中。用户可以通过 设置 hive.mapjoin.localtask.max.memory.usage = 0.999; 来更改此阈值
1.2 删除 JDBM
以前，Hive 使用JDBM(HIVE-1293)作为持久性哈希表。每当内存中的哈希表不能再保存数据时，它将把键/值交换到 JDBM 表中。但是，在分析 Map Join 时，我们发现这个 JDBM 组件花费了 70％以上的 CPU 时间，如图 3 所示。同样，生成的持久性文件 JDBM 太大，无法放入分布式缓存中。例如，如果用户将 67,000 个简单的整数键/值对放入 JDBM 中，它将生成更多的 22M 哈希表文件。因此，对于 Map Join 来说，JDBM 太重了，最好从 Hive 中删除此组件。 Map Join 旨在将小表的数据保存到内存中。如果表太大而无法容纳，只需作为通用联接运行。不再需要使用持久性哈希表。 (HIVE-1754)
图 3. JDBM 的分析结果
1.3 绩效评估
这是以前的 Map Join 与优化的 Map Join 之间的一些性能比较结果
表 1：以前的 Map 联接与新的优化的 Map 联接之间的比较
如表 1 所示，优化的 Map Join 将比上一个快 12~26 倍。Map 联接性能的大部分改进都来自于删除 JDBM 组件。
2.自动将联接转换为 Map 联接
2.1 新的联接执行流程
由于 Map 联接比普通联接快，因此最好在可能的情况下运行 Map 联接。以前，Hive 用户需要在查询中给出提示以指定小表是哪个表。例如，从 src1 x 选择 *** select/** * * mapjoin(a) */在 x.key = y.key *上加入 src2y;这不是用户体验和查询性能的好方法，因为有时用户可能会给出错误的提示，而用户也可能不会给出任何提示。在没有用户提示的情况下将 Common Join 转换为 Map Join 会更好。
(HIVE-1642)通过自动将通用联接转换为 Map 联接解决了该问题。对于 Map Join，查询处理器应该知道大表是哪个 Importing 表。在执行阶段，其他 Importing 表将被识别为小表，并且这些表需要保留在内存中。但是，通常，查询处理器不知道编译期间的 Importing 文件大小(即使有统计信息)，因为某些表可能是从子查询生成的中间表。因此，查询处理器只能在执行期间找出 Importing 文件的大小。
现在，用户需要通过 设置 hive.auto.convert.join = true; 启用此功能
这将是配置单元 0.11 中带有(HIVE-3297)的默认值
图 5：联接执行流程
如图 5 所示，左侧显示了先前的 Common Join 执行流程，这非常简单。另一方面，右侧是新的 Common Join 执行流程。在编译期间，查询处理器将生成一个条件任务，其中包含任务列表，这些任务之一将被解析为在执行期间运行。这意味着条件任务列表中的任务是候选任务，并且将选择其中一个在运行时运行。首先，应将原始的“普通加入任务”放入列表中。通过假定每个 Importing 表可能是大表，查询处理器还将生成一系列 Map Join Task。例如，从 src1 x 中选择 *** select 在 x.key = y.key* *上加入 src2y。 * _src2 _ 和 _src1 _ *这两个表都可能是大表，因此它将生成 2 Map Join Task。一个假设 src1 是大表，另一个假设 src2 是大表，如图 6 所示。
图 6 通过假设 Importing 表之一是大表来创建 Map 联接任务
2.2 在运行时解决联接操作
在执行阶段，条件任务可以准确知道每个 Importing 表的文件大小，即使该表是中间表也是如此。如果所有表都太大而无法转换为 Map 联接，则只需像以前一样运行 Common Join Task。如果其中一个表很大，而其他表又足够小，可以运行 Map Join，那么条件任务将选择相应的 Map Join 本地任务来运行。通过这种机制，它可以自动动态地将“普通联接”转换为“Map 联接”。
当前，如果小表的总大小大于 25M，则条件任务将选择原始的 Common Join 运行。 25M 是一个非常保守的数字，用户可以通过 *** set hive.mapjoin.smalltable.filesize = 30000000** *来更改此数字。
2.3 备份任务
如上所述，Map Join 的本地任务非常占用内存。因此，查询处理器将在子 jvm 中启动此任务，该子 jvm 具有与 Mapper 相同的堆大小。由于本地任务可能会用完内存，因此查询处理器将非常仔细地测量本地任务的内存使用情况。一旦本地任务的内存使用率高于阈值。此本地任务将自身中止，这意味着 Map 连接任务失败。在这种情况下，查询处理器将启动原始的 Common Join 任务作为要运行的备份任务，这对用户是完全透明的。基本思想如图 7 所示。
图 7，运行原始普通联接作为备份任务
2.4 绩效评估
这是以前的通用联接与优化的通用联接之间的一些性能比较结果。此处所有基准测试查询都可以转换为 Map Join。
表 2：以前的联接与新的优化联接之间的比较
*
对于先前的普通联接，实验仅计算 map 减少任务执行时间的平均时间。因为作业完成时间将包括作业调度开销。有时它将 await 一段时间才能开始在集群中运行作业。同样对于新的优化公共联接，该实验仅将本地任务执行时间的平均时间与 Map 的平均时间减少的执行时间相加。因此，这两个结果都避免了作业调度开销。
从结果来看，如果可以将新的普通联接转换为 Map 联接，则性能将提高 57％~163％。