对于Hive MAPJOIN作业，有多少数据被认为“太大”？

Question

编辑：添加了更多文件大小的详细信息，以及一些其他会话信息。

我有一个看似简单的Hive JOIN查询，令人惊讶地需要几个小时才能运行。

SELECT a.value1, a.value2, b.value
FROM a
JOIN b ON a.key = b.key
WHERE a.keyPart BETWEEN b.startKeyPart AND B.endKeyPart;

我正在尝试确定我的数据集和AWS硬件选择的执行时间是否正常，或者我是否只是尝试加入太多数据。

• 表A：~220万行，12MB压缩，81MB原始，4个文件。
• 表B：~24.5万行，6.7MB压缩，14MB原始，一个文件。
• AWS：emr-4.3.0，运行在大约5 m3.2xlarge EC2实例上。

来自A的记录总是匹配B中的一个或多个记录，因此从逻辑上讲，我看到在使用WHERE子句修剪之前最多生成了5000亿行。

为作业分配了4个映射器，在6小时内完成。这种查询和配置是正常的吗？如果没有，我该怎么做才能改善它？

我在JOIN键上划分了B，产生了5个分区，但没有注意到显着的改进。

此外，日志显示Hive优化器启动本地映射连接任务，可能是为了缓存或流式传输较小的表：

2016-02-07 02:14:13 Starting to launch local task to process map join;  maximum memory = 932184064
2016-02-07 02:14:16 Dump the side-table for tag: 1 with group count: 5 into file: file:/mnt/var/lib/hive/tmp/local-hadoop/hive_2016-02-07_02-14-08_435_7052168836302267808-1/-local-10003/HashTable-Stage-4/MapJoin-mapfile01--.hashtable
2016-02-07 02:14:17 Uploaded 1 File to: file:/mnt/var/lib/hive/tmp/local-hadoop/hive_2016-02-07_02-14-08_435_7052168836302267808-1/-local-10003/HashTable-Stage-4/MapJoin-mapfile01--.hashtable (12059634 bytes)
2016-02-07 02:14:17 End of local task; Time Taken: 3.71 sec.

是什么导致这项工作运行缓慢？数据集看起来不是太大，“小表”大小远低于25MB的“小表”限制，触发了MAPJOIN优化的禁用。

EXPLAIN输出的转储是copied on PasteBin以供参考。

我的会话支持输出和中间存储的压缩。这可能是罪魁祸首吗？

SET hive.exec.compress.output=true;
SET hive.exec.compress.intermediate=true;
SET mapred.output.compress=true;
SET mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;
SET io.compression.codecs=org.apache.hadoop.io.compress.GzipCodec;
SET io.seqfile.compression.type=BLOCK;

Answer 1

我对这个问题的解决方案是完全在JOIN ON子句中表达JOIN谓词，因为这是在Hive中执行JOIN的最有效方法。至于为什么原始查询很慢，我相信映射器只需要时间来逐行扫描中间数据集，超过100亿次。

由于Hive仅支持JOIN ON子句中的等式表达式并拒绝使用两个表别名作为参数的函数调用，因此无法将原始查询的BETWEEN子句重写为代数表达式。例如，以下表达式是非法的。

-- Only handles exclusive BETWEEN
JOIN b ON a.key = b.key
AND sign(a.keyPart - b.startKeyPart) = 1.0  -- keyPart > startKeyPart
AND sign(a.keyPart - b.endKeyPart) = -1.0   -- keyPart < endKeyPart

我最终修改了我的源数据，以包含Hive startKeyPart数据类型中endKeyPart和ARRAY<BIGINT>之间的每个值。

CREATE TABLE LookupTable
    key BIGINT,
    startKeyPart BIGINT,
    endKeyPart BIGINT,
    keyParts ARRAY<BIGINT>;

或者，我可以使用自定义Java方法在我的查询中内联生成此值; LongStream.rangeClosed()方法仅在Java 8中可用，它不是AWS emr-4.3.0中的Hive 1.0.0的一部分。

现在我在数组中有整个键空间，我可以使用LATERAL VIEW和explode()将数组转换为表，按如下方式重写JOIN。

WITH b AS
(
    SELECT key, keyPart, value
    FROM LookupTable
    LATERAL VIEW explode(keyParts) keyPartsTable AS keyPart
)
SELECT a.value1, a.value2, b.value
FROM a
JOIN b ON a.key = b.key AND a.keyPart = b.keyPart;

最终结果是，与相同硬件配置上的原始6小时相比，上述查询大约需要3分钟才能完成。