大数据之Hive 企业级调优 完整使用（第十章）

Hive 企业级调优 完整使用

• 一、执行计划（Explain）
• • 1）基本语法
  • 2）案例实操
  • • （1）查看下面这条语句的执行计划
    • （2）查看详细执行计划
• 二、Fetch 抓取
• • 1）案例实操
• 三、本地模式
• • 1）案例实操
  • • （1）关闭本地模式（默认是关闭的），并执行查询语句
    • （2）开启本地模式，并执行查询语句
• 四、表的优化
• • 1）小表大表 Join（MapJOIN）
  • 2）案例实操
  • • 1）需求介绍
    • 2）开启 MapJoin 参数设置
    • 3）MapJoin 工作机制
    • 4） 建大表、小表和 JOIN 后表的语句
    • 5）分别向大表和小表中导入数据
    • 6) 小表 JOIN 大表语句
    • 7）大表 JOIN 小表语句
  • 2) 大表 Join 大表
  • • 1）空 KEY 过滤
    • 2) 案例实操
    • • （1）配置历史服务器
      • （2）创建原始数据空 id 表
      • （3）分别加载原始数据和空 id 数据到对应表中
      • （4）测试不过滤空 id
      • （5）测试过滤空 id
  • 2）空 key 转换
  • • 1) 案例实操：
    • • （1）设置 5 个 reduce 个数
      • （2）JOIN 两张表
  • 3）SMB(Sort Merge Bucket join)
  • • （1）创建第二张大表
    • （2）创建分通表 1,桶的个数不要超过可用 CPU 的核数
    • （3）创建分通表 2,桶的个数不要超过可用 CPU 的核数
    • （4）设置参数
    • （5）测试

一、执行计划（Explain） 1）基本语法

EXPLAIN [EXTENDED | DEPENDENCY | AUTHORIZATION] query

2）案例实操 （1）查看下面这条语句的执行计划

没有生成 MR 任务的

hive (default)> explain select * from emp;

Explain
STAGE DEPENDENCIES:
 Stage-0 is a root stage
STAGE PLANS:
 Stage: Stage-0
 Fetch Operator
 limit: -1
 Processor Tree:
 TableScan
 alias: emp
 Statistics: Num rows: 1 Data size: 7020 Basic stats: COMPLETE 
Column stats: NONE
 Select Operator
 expressions: empno (type: int), ename (type: string), job 
(type: string), mgr (type: int), hiredate (type: string), sal (type: 
double), comm (type: double), deptno (type: int)
 outputColumnNames: _col0, _col1, _col2, _col3, _col4, _col5, 
_col6, _col7
 Statistics: Num rows: 1 Data size: 7020 Basic stats: COMPLETE 
Column stats: NONE
 ListSink

有生成 MR 任务的

hive (default)> explain select deptno, avg(sal) avg_sal from emp group by 
deptno;
Explain
STAGE DEPENDENCIES:
 Stage-1 is a root stage
 Stage-0 depends on stages: Stage-1
STAGE PLANS:
 Stage: Stage-1
 Map Reduce
 Map Operator Tree:
 TableScan
 alias: emp
 Statistics: Num rows: 1 Data size: 7020 Basic stats: COMPLETE 
Column stats: NONE
 Select Operator
 expressions: sal (type: double), deptno (type: int)
 outputColumnNames: sal, deptno
 Statistics: Num rows: 1 Data size: 7020 Basic stats: COMPLETE 
Column stats: NONE
 Group By Operator
 aggregations: sum(sal), count(sal)
 keys: deptno (type: int)
 mode: hash
 outputColumnNames: _col0, _col1, _col2
 Statistics: Num rows: 1 Data size: 7020 Basic stats:
 COMPLETE Column stats: NONE
 Reduce Output Operator
 key expressions: _col0 (type: int)
 sort order: +
 Map-reduce partition columns: _col0 (type: int)
 Statistics: Num rows: 1 Data size: 7020 Basic stats: 
COMPLETE Column stats: NONE
 value expressions: _col1 (type: double), _col2 (type: 
bigint)
 Execution mode: vectorized
 Reduce Operator Tree:
 Group By Operator
 aggregations: sum(VALUE._col0), count(VALUE._col1)
 keys: KEY._col0 (type: int)
 mode: mergepartial
 outputColumnNames: _col0, _col1, _col2
 Statistics: Num rows: 1 Data size: 7020 Basic stats: COMPLETE 
Column stats: NONE
 Select Operator
 expressions: _col0 (type: int), (_col1 / _col2) (type: double)
 outputColumnNames: _col0, _col1
 Statistics: Num rows: 1 Data size: 7020 Basic stats: COMPLETE 
Column stats: NONE
 File Output Operator
 compressed: false
 Statistics: Num rows: 1 Data size: 7020 Basic stats: COMPLETE 
Column stats: NONE
 table:
 input format: 
org.apache.hadoop.mapred.SequenceFileInputFormat
 output format: 
org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat
 serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
 Stage: Stage-0
 Fetch Operator
 limit: -1
 Processor Tree:
 ListSink

（2）查看详细执行计划

hive (default)> explain extended select * from emp;
hive (default)> explain extended select deptno, avg(sal) avg_sal from emp 
group by deptno;

二、Fetch 抓取

Fetch 抓取是指，Hive 中对某些情况的查询可以不必使用 MapReduce 计算。例如：SELECt 
* FROM employees;在这种情况下，Hive 可以简单地读取 employee 对应的存储目录下的文件，
然后输出查询结果到控制台。

在 hive-default.xml.template 文件中 hive.fetch.task.conversion 默认是 more，
老版本 hive默认是 minimal，该属性修改为 more 以后，在全局查找、字段查找、limit 查找等都
不走mapreduce。

 hive.fetch.task.conversion
 more
 
 Expects one of [none, minimal, more].
 Some select queries can be converted to single FETCH task minimizing 
latency.
 Currently the query should be single sourced not having any subquery 
and should not have any aggregations or distincts (which incurs RS), 
lateral views and joins.
 0. none : disable hive.fetch.task.conversion
 1. minimal : SELECT STAR, FILTER on partition columns, LIMIT only
 2. more : SELECT, FILTER, LIMIT only (support TABLESAMPLE and 
virtual columns)
 

1）案例实操

（1）把 hive.fetch.task.conversion 设置成 none，然后执行查询语句，都会执行 mapreduce
程序。

hive (default)> set hive.fetch.task.conversion=none;
hive (default)> select * from emp;
hive (default)> select ename from emp;
hive (default)> select ename from emp limit 3;

（2）把 hive.fetch.task.conversion 设置成 more，然后执行查询语句，如下查询方式都不
会执行 mapreduce 程序。

hive (default)> set hive.fetch.task.conversion=more;
hive (default)> select * from emp;
hive (default)> select ename from emp;
hive (default)> select ename from emp limit 3;

三、本地模式

大多数的 Hadoop Job 是需要 Hadoop 提供的完整的可扩展性来处理大数据集的。不过，
有时 Hive 的输入数据量是非常小的。在这种情况下，为查询触发执行任务消耗的时间可能
会比实际 job 的执行时间要多的多。对于大多数这种情况，Hive 可以通过本地模式在单台机
器上处理所有的任务。对于小数据集，执行时间可以明显被缩短。

用户可以通过设置 hive.exec.mode.local.auto 的值为 true，来让 Hive 在适当的时候自动
启动这个优化。

set hive.exec.mode.local.auto=true; //开启本地 mr
//设置 local mr 的最大输入数据量，当输入数据量小于这个值时采用 local mr 的方式，默认
为 134217728，即 128M
set hive.exec.mode.local.auto.inputbytes.max=50000000;
//设置 local mr 的最大输入文件个数，当输入文件个数小于这个值时采用 local mr 的方式，默
认为 4
set hive.exec.mode.local.auto.input.files.max=10;

1）案例实操 （1）关闭本地模式（默认是关闭的），并执行查询语句

hive (default)> select count(*) from emp group by deptno;

（2）开启本地模式，并执行查询语句

hive (default)> set hive.exec.mode.local.auto=true;
hive (default)> select count(*) from emp group by deptno;

四、表的优化 1）小表大表 Join（MapJOIN）

将 key 相对分散，并且数据量小的表放在 join 的左边，可以使用 map join 让小的维度表
先进内存。在 map 端完成 join。

实际测试发现：新版的 hive 已经对小表 JOIN 大表和大表 JOIN 小表进行了优化。小表放
在左边和右边已经没有区别。

2）案例实操 1）需求介绍

测试大表 JOIN 小表和小表 JOIN 大表的效率

2）开启 MapJoin 参数设置

（1）设置自动选择 Mapjoin

set hive.auto.convert.join = true; 默认为 true

（2）大表小表的阈值设置（默认 25M 以下认为是小表）

set hive.mapjoin.smalltable.filesize = 25000000;

3）MapJoin 工作机制

4） 建大表、小表和 JOIN 后表的语句

// 创建大表

create table bigtable(id bigint, t bigint, uid string, keyword string, 
url_rank int, click_num int, click_url string) row format delimited 
fields terminated by 't';

// 创建小表

create table smalltable(id bigint, t bigint, uid string, keyword string, 
url_rank int, click_num int, click_url string) row format delimited 
fields terminated by 't';

// 创建 join 后表的语句

create table jointable(id bigint, t bigint, uid string, keyword string, 
url_rank int, click_num int, click_url string) row format delimited 
fields terminated by 't';

5）分别向大表和小表中导入数据

hive (default)> load data local inpath '/opt/module/data/bigtable' into 
table bigtable;
hive (default)>load data local inpath '/opt/module/data/smalltable' into 
table smalltable;

6) 小表 JOIN 大表语句

insert overwrite table jointable
select b.id, b.t, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url
from smalltable s
join bigtable b
on b.id = s.id;

7）大表 JOIN 小表语句

insert overwrite table jointable
select b.id, b.t, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url
from bigtable b
join smalltable s
on s.id = b.id;

2) 大表 Join 大表 1）空 KEY 过滤

有时 join 超时是因为某些 key 对应的数据太多，而相同 key 对应的数据都会发送到相同
的 reducer 上，从而导致内存不够。此时我们应该仔细分析这些异常的 key，很多情况下，
这些 key 对应的数据是异常数据，我们需要在 SQL 语句中进行过滤。例如 key 对应的字段为
空，操作如下

2) 案例实操 （1）配置历史服务器

mapreduce.jobhistory.address
hadoop102:10020


 mapreduce.jobhistory.webapp.address
  hadoop102:19888

启动历史服务器

sbin/mr-jobhistory-daemon.sh start historyserver

查看 jobhistory

http://hadoop102:19888/jobhistory

（2）创建原始数据空 id 表

// 创建空 id 表
create table nullidtable(id bigint, t bigint, uid string, keyword string, 
url_rank int, click_num int, click_url string) row format delimited 
fields terminated by 't';

（3）分别加载原始数据和空 id 数据到对应表中

hive (default)> load data local inpath '/opt/module/data/nullid' into 
table nullidtable;

（4）测试不过滤空 id

hive (default)> insert overwrite table jointable select n.* from 
nullidtable n left join bigtable o on n.id = o.id;

（5）测试过滤空 id

hive (default)> insert overwrite table jointable select n.* from (select 
* from nullidtable where id is not null) n left join bigtable o on n.id = 
o.id;

2）空 key 转换

有时虽然某个 key 为空对应的数据很多，但是相应的数据不是异常数据，必须要包含在
join 的结果中，此时我们可以表 a 中 key 为空的字段赋一个随机的值，使得数据随机均匀地
分不到不同的 reducer 上。例如

1) 案例实操：

不随机分布空 null 值：

（1）设置 5 个 reduce 个数

set mapreduce.job.reduces = 5;

（2）JOIN 两张表

insert overwrite table jointable
select n.* from nullidtable n left join bigtable b on n.id = b.id;

结果：如下图所示，可以看出来，出现了数据倾斜，某些 reducer 的资源消耗远大于其
他 reducer。

随机分布空 null 值
（1）设置 5 个 reduce 个数
set mapreduce.job.reduces = 5;
（2）JOIN 两张表

insert overwrite table jointable
select n.* from nullidtable n full join bigtable o on 
nvl(n.id,rand()) = o.id;

结果：如下图所示，可以看出来，消除了数据倾斜，负载均衡 reducer 的资源消耗

3）SMB(Sort Merge Bucket join) （1）创建第二张大表

create table bigtable2(
 id bigint,
 t bigint,
 uid string,
 keyword string,
 url_rank int,
 click_num int,
 click_url string)
row format delimited fields terminated by 't';
load data local inpath '/opt/module/data/bigtable' into table bigtable2;

测试大表直接 JOIN

insert overwrite table jointable
select b.id, b.t, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url
from bigtable s
join bigtable2 b
on b.id = s.id;

（2）创建分通表 1,桶的个数不要超过可用 CPU 的核数

create table bigtable_buck1(
 id bigint,
 t bigint,
 uid string,
 keyword string,
 url_rank int,
 click_num int,
 click_url string)
clustered by(id) 
sorted by(id)
into 6 buckets
row format delimited fields terminated by 't';
load data local inpath '/opt/module/data/bigtable' into table 
bigtable_buck1;

（3）创建分通表 2,桶的个数不要超过可用 CPU 的核数

 id bigint,
 t bigint,
 uid string,
 keyword string,
 url_rank int,
 click_num int,
 click_url string)
clustered by(id)
sorted by(id) 
into 6 buckets
row format delimited fields terminated by 't';
load data local inpath '/opt/module/data/bigtable' into table 
bigtable_buck2;

（4）设置参数

set hive.optimize.bucketmapjoin = true;
set hive.optimize.bucketmapjoin.sortedmerge = true;
set 
hive.input.format=org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat;

（5）测试

insert overwrite table jointable
select b.id, b.t, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url
from bigtable_buck1 s
join bigtable_buck2 b
on b.id = s.id;