一例 Hive join 优化实战

Posted on | Edited on | In | Views:

由于 hive 与传统关系型数据库面对的业务场景及底层技术架构都有着很大差异，因此，传统数据库领域的一些技能放到 Hive 中可能已不再适用。关于 hive 的优化与原理、应用的文章，前面也陆陆续续的介绍了一些，但大多都偏向理论层面，本文就介绍一个实例，从实例中一步步加深对 hive 调优的认识与意识。

1、需求

需求我做了简化，很简单，两张表做个 join，求指定城市，每天的 pv，用传统的 RDBMS SQL 写出来就这样的：

1
2
3
4
5
6
7
8
9
10
11
SELECT t.statdate,
       c.cname,
       count(t.cookieid)
FROM tmpdb.city c
JOIN ecdata.ext_trackflow t ON (t.area1= c.cname
                                OR t.area2 =c.cname
                                OR t.area3 = c.cname)
WHERE t.statdate>='20140818' and t.statdate<='20140824'
  AND platform='pc'
GROUP BY t.statdate,
         c.cname;

怎么样？根据 SQL 看懂需求没问题吧？

2、非等值 join 问题

然后把这条 SQL 贴到 hive 中去执行，然后你会发现报错了：

1
FAILED: SemanticException [Error 10019]: Line 5:32 OR not supported in JOIN currently 'cname'

这是因为 hive 受限于 MapReduce 算法模型，只支持 equi-joins（等值 join），要实现上述的非等值 join，你可以采用笛卡儿积（ full Cartesian product ）来实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
SELECT t.statdate,
       c.cname,
       count(t.cookieid)
FROM tmpdb.city c
JOIN ecdata.ext_trackflow t
WHERE t.statdate>='20140818'
  AND t.statdate<='20140824'
  AND platform='pc'
  AND (t.area1= c.cname
       OR t.area2 =c.cname
       OR t.area3 = c.cname)
GROUP BY t.statdate,
         c.cname;

然后再拿着这条语句执行下。

3、优化：reduce side join VS Cartesian product

如果你真的把这条语句放到 Hive 上执行，然后恰好你有张表还非常大，那么恭喜你。。。集群管理员估计会找你的麻烦了。。。

友情提示：笛卡儿积这种语句在 Hive 下慎用，大数据场景下的 m * n 映射结果你懂的。。。对此，Hive 特意提供了一个环境变量：hive.mapred.mode=strict; 防止笛卡儿积的执行：

1
FAILED: SemanticException [Error 10052]: In strict mode, cartesian product is not allowed. If you really want to perform the operation, set hive.mapred.mode=nonstrict

从 2 中的观察得知我们在 on 后面跟 join 条件，走的是 reduce side join，如果你在 where 后跟则是走 Cartesian product，但是这里单条 sql 又没法实现 reduce side join，还有没有其它办法呢？

4、改写非等值 join：union all

既然不允许非等值 join，那我们换一下思路，多个子查询 union all，然后汇总：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
SELECT dt,
       name,
       count(cid)
FROM
  (SELECT t.statdate dt,
          c.cname name,
          t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area1 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc'
   UNION ALL SELECT t.statdate dt,
                    c.cname name,
                    t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area2 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc'
   UNION ALL SELECT t.statdate dt,
                    c.cname name,
                    t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area3 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc') tmp_trackflow
GROUP BY dt,
         name;

5、优化：map side join

上述语句走的是 reduce side join，从我们的需求及业务得知，tmpdb.city 是一张字典表，数据量很小，因此我们可以试试把上述的语句改写成 mapjoin：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
SELECT dt,
       name,
       count(cid)
FROM
  (SELECT /*+ MAPJOIN(c) */ t.statdate dt,
                            c.cname name,
                            t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area1 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc'
   UNION ALL SELECT /*+ MAPJOIN(c) */ t.statdate dt,
                                      c.cname name,
                                      t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area2 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc'
   UNION ALL SELECT /*+ MAPJOIN(c) */ t.statdate dt,
                                      c.cname name,
                                      t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area3 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc') tmp_trackflow
GROUP BY dt,
         name;

6、优化无极限：开启 parallel 和 控制 reduce 个数

上述语句执行时，你可以看到执行计划和状态信息，以及结合你的 union all 语句可知，三个 union 语句之间没有依赖关系，其实是可以并行执行的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
explain SQL...
...
STAGE DEPENDENCIES:
  Stage-11 is a root stage
  Stage-1 depends on stages: Stage-11
  Stage-2 depends on stages: Stage-1
  Stage-3 depends on stages: Stage-2, Stage-6, Stage-9
  Stage-12 is a root stage
  Stage-5 depends on stages: Stage-12
  Stage-6 depends on stages: Stage-5
  Stage-13 is a root stage
  Stage-8 depends on stages: Stage-13
  Stage-9 depends on stages: Stage-8
  Stage-0 is a root stage
...

我们在 SQL 前加上如下环境变量选项：

1
2
set mapred.reduce.tasks=60;
set hive.exec.parallel=true;

让执行计划中的 Stage-11、Stage-12、Stage-13 并行执行，并控制好 reduce task 个数。

完整的语句如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
hive -e "
SET mapred.reduce.tasks=60;


SET hive.exec.parallel=TRUE;


SELECT dt,
       name,
       count(cid)
FROM
  (SELECT /*+ MAPJOIN(c) */ t.statdate dt,
                            c.cname name,
                            t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area1 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc'
   UNION ALL SELECT /*+ MAPJOIN(c) */ t.statdate dt,
                                      c.cname name,
                                      t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area2 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc'
   UNION ALL SELECT /*+ MAPJOIN(c) */ t.statdate dt,
                                      c.cname name,
                                      t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area3 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc') tmp_trackflow
GROUP BY dt,
         name;

" > a1.txt

最后的优化效果是：2 中的语句三个小时没出结果。。。5 比 4 快 8 倍左右，6 比 5 快 2 倍左右，最终 10min 出结果。

7、最后的问题：

在 6 的语句执行的时候你会发现，其扫描了 三遍 源文件。而 hive 本身是对 union all 的 join 做了优化的，当多个 union all 子查询同一张表时，只扫描一次源文件，但这里为什么会三个子查询各扫描一次呢？

可能是这里的 union all 子查询使用了 join 的缘故，导致 hive 的 union all 执行计划优化失效了。

关于这块怎么能优化成只扫描一次源文件，或者你有更好的优化方案，欢迎留言交流。

8、关于 hive 中的 笛卡尔集（ full Cartesian product ）

在JION接连查询中没有ON连接key，而通过WHERE条件语句会产生笛卡尔集。

Hive本身是不支持笛卡尔集的，不能用select T1., T2. from table1, table2这种语法。但有时候确实需要用到笛卡尔集的时候，可以用下面的语法来实现同样的效果：

1
select T1.*, T2.* from table1 T1 join table2 T2 where 1=1;

注意在Hive的Strict模式下不能用这种语法，因为这样会产生笛卡尔集，而这种模式禁止产生笛卡尔集。需要先用set hive.mapred.mode=nonstrict;设为非strict模式就可以用了，或者将where改为on连接。

1
select T1.*, T2.* from table1 T1 join table2 T2 on  T1.id=T2.id;

9、关于Strict Mode

Hive中的严格模式可以防止用户发出（可以有问题）的查询无意中造成不良的影响。 将hive.mapred.mode设置成strict可以禁止三种类型的查询：

1）、在一个分区表上，如果没有在WHERE条件中指明具体的分区，那么这是不允许的，换句话说，不允许在分区表上全表扫描。这种限制的原因是分区表通常会持非常大的数据集并且可能数据增长迅速，对这样的一个大表做全表扫描会消耗大量资源，必须要再WHERE过滤条件中具体指明分区才可以执行成功的查询。

2）、第二种是禁止执行有ORDER BY的排序要求但没有LIMIT语句的HiveQL查询。因为ORDER BY全局查询会导致有一个单一的reducer对所有的查询结果排序，如果对大数据集做排序，这将导致不可预期的执行时间，必须要加上limit条件才可以执行成功的查询。

3）、第三种是禁止产生笛卡尔集。在JION接连查询中没有ON连接key而通过WHERE条件语句会产生笛卡尔集，需要改为JOIN…ON语句。

10、Refer：

[1] Hive Query- Joining two tables on three joining conditions with OR operator

http://stackoverflow.com/questions/16272804/hive-query-joining-two-tables-on-three-joining-conditions-with-or-operator

[2] LanguageManual JoinOptimization

https://cwiki.apache.org/confluence/display/Hive/LanguageManual+JoinOptimization

[3] hive 执行计划

http://yychao.iteye.com/blog/1749562

[4] Hive SQL解析/执行计划生成流程分析

http://yanbohappy.sinaapp.com/?p=265

[5] 数据仓库中的SQL性能优化（Hive篇）

http://www.zihou.me/html/2014/02/12/9207.html

[6] Hive优化以及执行原理

http://www.smartcitychina.cn/upload/2014-01/14012015376829.pdf

[7] Hive作业优化总结

http://my.oschina.net/yangzhiyuan/blog/262910

[8] Hive连接产生笛卡尔集

http://blog.javachen.com/2013/10/17/cartesian-product-in-hive-inner-join/#