一例 Hive join 优化实战

时间 2019-11-11

标签一例 hive join 优化实战栏目 Hadoop 繁體版

原文原文链接

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

一、需求

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

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 看懂需求没问题吧？

二、非等值 join 问题

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

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

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

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;

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

三、优化：reduce side join VS Cartesian product

若是你真的把这条语句放到 Hive 上执行，而后刚好你有张表还很是大，那么恭喜你。。。集群管理员估计会找你的麻烦了。。。 sql

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

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，还有没有其它办法呢？ apache

四、改写非等值 join：union all

既然不容许非等值 join，那咱们换一下思路，多个子查询 union all，而后汇总：性能优化

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;

五、优化：map side join

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

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;

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

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

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 前加上以下环境变量选项：

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

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

完整的语句以下： app

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 出结果。

七、最后的问题：

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

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

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

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

在JION接连查询中没有ON链接key，而经过WHERE条件语句会产生笛卡尔集。
Hive自己是不支持笛卡尔集的，不能用select T1.*, T2.* from table1, table2这种语法。但有时候确实须要用到笛卡尔集的时候，能够用下面的语法来实现一样的效果：
select T1.*, T2.* from table1 T1 join table2 T2 where 1=1;
注意在Hive的Strict模式下不能用这种语法，由于这样会产生笛卡尔集，而这种模式禁止产生笛卡尔集。须要先用set hive.mapred.mode=nonstrict;设为非strict模式就能够用了，或者将where改成on链接。
select T1.*, T2.* from table1 T1 join table2 T2 on T1.id=T2.id;

九、关于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语句。