Hive 交叉连接在本地地图连接上失败

Question

是否有解决以下错误的直接方法或总体上更好的方法来使用 Hive 获得我需要的连接？不需要输出到存储表，因为我可以满足于将INSERT OVERWRITE LOCAL DIRECTORY 转换为 csv。

我正在尝试执行以下交叉连接。 ipint 是一个 9GB 的表，geoiplite 是 270MB。

CREATE TABLE iplatlong_sample AS
SELECT ipintegers.networkinteger, geoiplite.latitude, geoiplite.longitude
FROM geoiplite
CROSS JOIN ipintegers
WHERE ipintegers.networkinteger >= geoiplite.network_start_integer AND ipintegers.networkinteger <= geoiplite.network_last_integer;

我在 ipintegers 上使用 CROSS JOIN 而不是 geoiplite，因为我读过规则是较小的表在左侧，较大的在右侧。

根据 HIVE，Map 和 Reduce 阶段完成到 100%，但随后

2015-08-01 04:45:36,947 第一阶段地图 = 100%，减少 = 100%，累积 CPU 8767.09 秒

MapReduce 总累计 CPU 时间：0 天 2 小时 26 分 7 秒 90 毫秒

已结束的工作 = job_201508010407_0001

Stage-8 由条件解析器选择。

执行日志位于：/tmp/myuser/.log

2015-08-01 04:45:38 开始启动本地任务来处理地图 加入;最大内存 = 12221153280

执行失败，退出状态：3

获取错误信息

任务失败！

任务 ID：第 8 阶段

日志：

/tmp/myuser/hive.log

FAILED：执行错误，返回代码 3 从 org.apache.hadoop.hive.ql.exec.mr.MapredLocalTask​​

MapReduce 作业 已启动：作业 0：地图：38 减少：1 累积 CPU：8767.09 秒
HDFS 读取：9438495086 HDFS 写入：8575548486 成功

我的蜂巢配置：

SET hive.mapred.local.mem=40960;
SET hive.exec.parallel=true;
SET hive.exec.compress.output=true;
SET hive.exec.compress.intermediate = true;
SET hive.optimize.skewjoin = true;
SET mapred.compress.map.output=true;
SET hive.stats.autogather=false;

我在 true 和 false 之间更改了 SET hive.auto.convert.join，但结果相同。

这是来自 /tmp/myuser/hive.log 的输出日志中的错误

$ tail -12 -f tmp/mysyer/hive.log

2015-08-01 07:30:46,086 ERROR exec.Task (SessionState.java:printError(419)) - Execution failed with exit status: 3
2015-08-01 07:30:46,086 ERROR exec.Task (SessionState.java:printError(419)) - Obtaining error information
2015-08-01 07:30:46,087 ERROR exec.Task (SessionState.java:printError(419)) -
Task failed!
Task ID:
  Stage-8

Logs:

2015-08-01 07:30:46,087 ERROR exec.Task (SessionState.java:printError(419)) - /tmp/myuser/hive.log
2015-08-01 07:30:46,087 ERROR mr.MapredLocalTask (MapredLocalTask.java:execute(268)) - Execution failed with exit status: 3
2015-08-01 07:30:46,094 ERROR ql.Driver (SessionState.java:printError(419)) - FAILED: Execution Error, return code 3 from org.apache.hadoop.hive.ql.exec.mr.MapredLocalTask

我在 Master 上运行 hive 客户端，这是一个 n1-highmem-8 类型（8 CPU，52GB）的 Google Cloud Platform 实例，worker 是 n1-highmem-4（4CPU 26GB），但我怀疑在 MAP 之后并减少在 Master 上发生本地连接（如暗示的那样）。无论如何，在 bdutils 中，我将工作节点 (n1-highmem-4) 的 JAVAOPTS 配置为：n1-highmem-4

解决方案编辑：解决方案是将数据范围数据组织成范围树。

Answer 1

我认为不可能执行这种交叉连接蛮力 - 只需将行号相乘，这有点失控。您需要一些优化，我认为 hive 尚不具备这些功能。

但是这个问题实际上可以在 O(N1+N2) 时间内解决吗，前提是您已对数据进行了排序（hive 可以为您完成） - 您只需同时浏览两个列表，每一步都获得一个 ip 整数，查看是否有任何间隔从这个整数开始，添加它们，删除那些结束的，发出匹配的元组，等等。伪代码：

intervals=[]
ipintegers = iterator(ipintegers_sorted_file)
intervals = iterator(intervals_sorted_on_start_file)
for x in ipintegers:
    intervals = [i for i in intervals if i.end >= x]

    while(intervals.current.start<=x):
        intervals.append(intervals.current)
        intervals.next()
    for i in intervals:
        output_match(i, x)

现在，如果您有一个外部脚本/UDF 函数，该函数知道如何读取较小的表并获取 ip 整数作为输入并将匹配的元组作为输出，您可以使用 hive 和 SELECT TRANSFORM 将输入流式传输到它。

或者您可以只在具有两个输入文件的本地计算机上运行此算法，因为这只是 O(N)，甚至 9 GB 的数据也是非常可行的。