Join Optimization

有关 Hive 联接的一般讨论，包括语法，示例和限制，请参阅Joins Wiki 文档。

Hive Optimizer 的改进

本文档介绍了 Hive 查询执行计划的优化，以提高联接效率并减少对用户提示的需求。

Hive 会自动识别各种用例并对其进行优化。 Hive 0.11 改进了以下情况的优化程序：

• 在一侧适合内存的位置连接。在新的优化中：

• 将该那一侧作为哈希表加载到内存中

  • 只需要扫描较大的表

  • 事实表在内存中的占用空间较小

• Star-schema joins

• 在许多情况下，不再需要提示。

• Map 联接由优化器自动拾取。

星空加入优化

决策支持系统或数据仓库的简单模式是星型模式，其中事件收集在大型事实表中，而较小的支持表(维度)用于描述数据。

TPC DS是此类架构的示例。它为典型的零售仓库建模，事件是销售，典型尺寸是销售日期，销售时间或购买方的人口统计。典型查询沿维度表中的属性汇总和过滤事实表。

星型模式示例

Select count(*) cnt
From store_sales ss
     join household_demographics hd on (ss.ss_hdemo_sk = hd.hd_demo_sk)
     join time_dim t on (ss.ss_sold_time_sk = t.t_time_sk)
     join store s on (s.s_store_sk = ss.ss_store_sk)
Where
     t.t_hour = 8
     t.t_minute >= 30
     hd.hd_dep_count = 2
order by cnt;

先前对 MAPJOIN 的支持

Hive 支持 MAPJOIN，它们非常适合这种情况–至少对于足够小以适合内存的尺寸而言。在 0.11 版之前，可以通过优化器提示调用 MAPJOIN：

select /*+ MAPJOIN(time_dim) */ count(*) from
store_sales join time_dim on (ss_sold_time_sk = t_time_sk)

或通过自动加入转换：

set hive.auto.convert.join=true;
select count(*) from
store_sales join time_dim on (ss_sold_time_sk = t_time_sk)

hive.auto.convert.join的默认值在 Hive 0.10.0 中为 false。 Hive 0.11.0 将默认值更改为 true(HIVE-3297)。请注意hive-default.xml.template错误地将 Hive 0.11.0 到 0.13.1 中的默认值设置为 false。

通过将较小的表加载到内存中的哈希 Map 中，并在流传输时将键与较大的表进行匹配，来处理 MAPJOIN。先前的实现具有以下分工：

• Local work:

• 通过标准表扫描(包括过滤器和投影)从本地计算机上的源读取记录

  • 在内存中构建哈希表

  • 将哈希表写入本地磁盘

  • 将哈希表上传到 dfs

  • 将哈希表添加到分布式缓存

• Map task

• 从本地磁盘(分布式缓存)读取哈希表到内存

  • 将记录的键与哈希表匹配

  • 组合匹配并写入输出

• 没有减少任务

先前实施的限制

Hive 0.11 之前的 MAPJOIN 实现具有以下限制：

• mapjoin 运算符一次只能处理一个键。也就是说，它可以执行多表联接，但前提是所有表都在同一键上联接。 (典型的星型模式联接不属于此类别.)

• 要使用户正确应用提示很麻烦，并且自动转换没有足够的逻辑来一致地预测 MAPJOIN 是否适合内存。

• 除非将查询写为mapjoin(table, subquery(mapjoin(table, subquery....)的级联序列，否则不会将 MAPJOIN 链合并为一个仅 Map 的作业。自动转换永远不会产生仅 Map 作业。

• 必须为查询的每次运行生成 mapjoin 运算符的哈希表，这涉及将所有数据下载到 HiveClient 端计算机以及上载生成的哈希表文件。

星空加入的增强功能

Hive 0.11 中的优化器增强功能集中在星型架构配置中所需的联接的有效处理上。最初的工作仅限于星型模式联接，其中所有经过过滤和投影的维表同时适合内存。现在，也只实现了一些维表适合内存的方案(HIVE-3996)。

联接优化可以分为三个部分：

• 使用 maphints 时，在单个仅 Map 作业中在运算符树中执行 mapjoin 链。

• 将优化扩展到自动转换的情况(优化时生成适当的备份计划)。

• 完全在任务侧生成内存中的哈希表。 (Future 的工作.)

以下各节介绍了这些优化器增强功能。

优化 Map 联接链

以下查询在执行时将产生两个单独的仅 Map 作业：

select /*+ MAPJOIN(time_dim, date_dim) */ count(*) from
store_sales 
join time_dim on (ss_sold_time_sk = t_time_sk) 
join date_dim on (ss_sold_date_sk = d_date_sk)
where t_hour = 8 and d_year = 2002

但是，对于小尺寸表，可能需要将两个表的一部分同时放入内存。由于事实表仅被读取一次，而不是被读取两次并将其写入 HDFS 以在作业之间进行通信，因此大大减少了执行该查询所需的时间。

当前和将来的优化

• 将 M * -MR 模式合并为一个 MR。

• 尽可能将 MJ-> MJ 合并为一个 MJ。

• 将 MJ *模式作为 MJ 运算符链合并到一个 Map 阶段。 (尚未实现.)

如果hive.auto.convert.join设置为 true，则优化器不仅将联接转换为 mapjoin，而且还尽可能合并 MJ *模式。

优化自动加入转换

启用自动联接后，不再需要在查询中提供 Map 联接提示。可以使用两个配置参数启用自动加入选项：

set hive.auto.convert.join.noconditionaltask = true;
set hive.auto.convert.join.noconditionaltask.size = 10000000;

hive.auto.convert.join.noconditionaltask的默认值为 true，表示启用了自动转换。 (最初的默认值是 false –请参见HIVE-3784 –但在发布 Hive 0.11.0 之前，它已由HIVE-4146更改为 true。)

size configuration使用户可以控制内存中可以容纳什么大小的表。此值表示可以转换为适合内存的哈希表的表大小的总和。当前，联接的 n-1 个表必须容纳在内存中才能使 Map 联接优化生效。没有检查表是否为压缩表，表的潜在大小为多少。下一节将讨论此假设对结果的影响。

例如，上一个查询变为：

select count(*) from
store_sales 
join time_dim on (ss_sold_time_sk = t_time_sk)
join date_dim on (ss_sold_date_sk = d_date_sk)
where t_hour = 8 and d_year = 2002

如果 time_dim 和 date_dim 适合提供的大小配置，则将相应的联接转换为 map-joins。如果表大小的总和可适合配置的大小，则将两个 Map 联接组合在一起，从而形成单个 Map 联接。这减少了所需的 MR 作业数量，并大大提高了此查询的执行速度。此示例也可以轻松扩展为多向联接，并且将按预期工作。

外连接带来更多挑战。由于 map-join 运算符只能流化一个表，因此流化的表必须是需要所有行的表。对于左外部联接，这是联接左侧的表；对于右侧的外部联接，右侧的表等。这意味着即使可以将内部联接转换为 Map 联接，也不能转换外部联接。仅当表(需要流传输的表除外)可以适合大小配置时，才能转换外部联接。完全无法将完全外部联接转换为 Map 联接，因为这两个表都需要进行流处理。

自动联接转换还会影响 sort-merge-bucket 联接。

Current Optimization

• 将尽可能多的 MJ 操作员分组为一个 MJ。

当 Hive 根据配置标志进行联接运算符的 Map 联接转换时，在这些转换结束时会尽力将尽可能多的分组在一起。按 Sequences 进行操作，如果参与各个 map-join 运算符的表的总和在noConditionalTask.size标志配置的限制内，则将这些 MJ 运算符组合在一起。这样可以确保这些查询的速度更快。

自动转换为 SMB Map Join

排序合并桶(SMB)联接也可以转换为 SMBMap 联接。 SMB 联接用于对表进行排序和存储的任何位置。联接归结为仅合并已排序的表，从而使该操作比普通的 Map 联接更快。但是，如果对表进行了分区，则速度可能会变慢，因为每个 Map 器都需要获得只有一个键的分区的很小一部分。

以下配置设置允许将 SMB 转换为 Map 联接 SMB：

set hive.auto.convert.sortmerge.join=true;
set hive.optimize.bucketmapjoin = true;
set hive.optimize.bucketmapjoin.sortedmerge = true;

有一个选项可以使用以下配置来设置大表选择策略：

set hive.auto.convert.sortmerge.join.bigtable.selection.policy 
    = org.apache.hadoop.hive.ql.optimizer.TableSizeBasedBigTableSelectorForAutoSMJ;

默认情况下，选择策略是平均分区大小。与哈希和流传输相比，大表选择策略有助于确定只为流传输选择哪个表。

可用的选择策略是：

org.apache.hadoop.hive.ql.optimizer.AvgPartitionSizeBasedBigTableSelectorForAutoSMJ (default)
org.apache.hadoop.hive.ql.optimizer.LeftmostBigTableSelectorForAutoSMJ
org.apache.hadoop.hive.ql.optimizer.TableSizeBasedBigTableSelectorForAutoSMJ

名称描述了它们的用途。这对于事实联接(TPC DS基准中的查询 82)特别有用。

SMB 使用不同的键跨表联接

如果表的键数不同，例如表 A 有 2 个 SORT 列，而表 B 有 1 个 SORT 列，则可能会导致索引超出范围异常。

以下查询导致索引超出范围异常，因为 emp_person 可以说例如具有 1 个排序列，而 emp_pay_history 具有 2 个排序列。

错误配置单元 0.11

SELECT p.*, py.*
FROM emp_person p INNER JOIN emp_pay_history py
ON   p.empid = py.empid

这很好。

工作查询 Hive 0.11

SELECT p.*, py.*
FROM emp_pay_history py INNER JOIN emp_person p
ON   p.empid = py.empid

在任务端生成哈希表

将来的工作将使在任务侧完全生成内存中的哈希表成为可能。

Client 端哈希表的优缺点

在 Client 端计算机上生成哈希表(或用于多表联接的多个哈希表)存在缺陷。 (Client 端计算机是用于运行 HiveClient 端和提交作业的主机.)

• 数据位置： Client 端计算机通常不是数据节点。所有访问的数据都是远程的，必须通过网络读取。

• 规格： 出于相同的原因，目前尚不清楚运行此处理的机器的规格。它可能具有任务节点所没有的内存，硬盘驱动器或 CPU 限制。

• HDFS 上载： 数据必须带回到群集并通过分布式缓存复制，以供任务节点使用。

在 Client 端计算机上预处理哈希表也有一些好处：

• 分布式缓存中存储的内容可能小于原始表(过滤器和投影)。

• 相反，使用分布式缓存将哈希表直接加载到任务节点上意味着缓存中的对象更大，从而潜在地减少了使用 MAPJOIN 的机会。

哈希表的任务侧生成

当在任务侧完全生成哈希表时，所有任务节点都必须访问原始数据源以生成哈希表。由于在正常情况下这将并行发生，因此不会影响延迟，但是 Hive 具有存储处理程序的概念，并且让许多任务访问同一外部数据源(HBase，数据库等)可能会使该源不堪重负或减慢其速度。

其他优化选项

• 增加维度表上的复制因子。

• 使用分布式缓存来保存维表。