大数据技术之Hive企业级调优Hive实战

Posted @从一到无穷大

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了大数据技术之Hive企业级调优Hive实战相关的知识，希望对你有一定的参考价值。

文章目录

1 企业级调优

1.1 执行计划（Explain）

1. 基本语法
EXPLAIN [EXTENDED | DEPENDENCY | AUTHORIZATION] query

2. 案例实操
（1）查看下面这条语句的执行计划
没有生成 MR 任务的

hive (default)> explain select * from emp;
OK
Explain
Plan optimized by CBO.

Stage-0
  Fetch Operator
    limit:-1
    Select Operator [SEL_1]
      Output:["_col0","_col1","_col2","_col3","_col4","_col5","_col6","_col7"]
      TableScan [TS_0]
        Output:["empno","ename","job","mgr","hiredate","sal","comm","deptno"]

Time taken: 3.189 seconds, Fetched: 10 row(s)

有生成 MR 任务的

hive (default)> explain select deptno, avg(sal) avg_sal from emp group by deptno;
OK
Explain
Plan optimized by CBO.

Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE)

Stage-0
  Fetch Operator
    limit:-1
    Stage-1
      Reducer 2 vectorized
      File Output Operator [FS_12]
        Select Operator [SEL_11] (rows=1 width=7030)
          Output:["_col0","_col1"]
          Group By Operator [GBY_10] (rows=1 width=7030)
            Output:["_col0","_col1","_col2"],aggregations:["sum(VALUE._col0)","count(VALUE._col1)"],keys:KEY._col0
          <-Map 1 [SIMPLE_EDGE] vectorized
            SHUFFLE [RS_9]
              PartitionCols:_col0
              Group By Operator [GBY_8] (rows=1 width=7030)
                Output:["_col0","_col1","_col2"],aggregations:["sum(sal)","count(sal)"],keys:deptno
                Select Operator [SEL_7] (rows=1 width=7030)
                  Output:["sal","deptno"]
                  TableScan [TS_0] (rows=1 width=7030)
                    default@emp,emp,Tbl:COMPLETE,Col:NONE,Output:["sal","deptno"]

Time taken: 1.893 seconds, Fetched: 25 row(s)

（2）查看详细执行计划

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

1.2 Fetch 抓取

Fetch抓取是指 ，Hive中对某些情况的查询可以不必使用 MapReduce计算 。例如 SELECT * FROM employees;在这种情况下 Hive 可以简单地读取 employee对应的存储目录下的文件然后输出查询结果到控制台。在hive-default.xml.template文件中 hive.fetch.task.conversion默认是 more，老版本 hive默认是minimal，该属性修改为 more以后，在全局查找、字段查找、 limit查找等都不走 mapreduce。

<property>
	<name>hive.fetch.task.conversion</name>
	<value>more</value>
	<description>
		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 haveany 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)
	</description>
</property>

案例实操：
（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;

1.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）关闭本地模式 （默认是关闭的 ），并执行查询语句

hive (default)> select count(*) from emp group by deptno;
OK
_c0
3
5
6
1
Time taken: 8.67 seconds, Fetched: 4 row(s)

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

hive (default)> set hive.exec.mode.local.auto=true;
hive (default)> select count(*) from emp group by deptno;
OK
_c0
3
5
6
1
Time taken: 9.457 seconds, Fetched: 4 row(s)

1.4 表的优化

1.4.1 小表大表Join（MapJOIN）

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

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

1. 需求介绍
测试大表JOIN 小表和小表JOIN 大表的效率。

2. 开启 MapJoin 参数设置
（1）设置自动选择 Mapjoin， 默认为true

hive (default)> set hive.auto.convert.join = true;

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

hive (default)> set hive.mapjoin.smalltable.filesize = 25000000;

3. MapJoin 工作机制

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

// 创建大表
hive (default)> 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';
OK
Time taken: 0.193 seconds

// 创建小表
hive (default)> 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';
OK
Time taken: 0.162 seconds

// 创建 join 后的表
hive (default)> 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';
OK
Time taken: 0.127 seconds

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

hive (default)> load data local inpath '/opt/module/hive-3.1.2/data/bigtable' into table bigtable;
Loading data to table default.bigtable
OK
Time taken: 27.455 seconds

hive (default)> load data local inpath '/opt/module/hive-3.1.2/data/smalltable' into table smalltable;
Loading data to table default.smalltable
OK
Time taken: 1.51 seconds

6. 小表 JOIN 大表语句

hive (default)> 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;
Loading data to table default.jointable
OK
b.id	b.t	b.uid	b.keyword	b.url_rank	b.click_num	b.click_url
Time taken: 97.527 seconds

7. 大表 JOIN小表语句

hive (default)> 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;
Loading data to table default.jointable
OK
b.id	b.t	b.uid	b.keyword	b.url_rank	b.click_num	b.click_url
Time taken: 94.968 seconds

1.4.2 大表 Join 大表

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

（1）配置历史服务器
配置mapred site.xml

<!--历史服务器端地址-->
    <property>
        <name>mapreduce.jobhistory.address</name>
        <value>hadoop102:10020</value>
    </property>
<!--历史服务器web端地址-->
    <property>
        <name>mapreduce.jobhistory.webapp.address</name>
        <value>hadoop102:19888</value>
    </property>

启动历史服务器

sbin/mr-jobhistory daemon.sh start historyserver

查看jobhistory
http://hadoop102:19888/jobhistory

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

hive (default)> 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';
OK
Time taken: 0.151 seconds

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

hive (default)> load data local inpath '/opt/module/hive-3.1.2/data/nullid' into table nullidtable;
Loading data to table default.nullidtable
OK
Time taken: 11.346 seconds

（4）测试不过滤空id

hive (default)> insert overwrite table jointable select n.* from
              > nullidtable n left join bigtable o on n.id = o.id;
OK
n.id	n.t	n.uid	n.keyword	n.url_rank	n.click_num	n.click_url
Time taken: 182.221 seconds

（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;
OK
n.id	n.t	n.uid	n.keyword	n.url_rank	n.click_num	n.click_url
Time taken: 152.913 seconds

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

不随机分布空 null 值：
（1）设置 5 个 reduce 个数

hive (default)> set mapreduce.job.reduces = 5;

（2）JOIN 两张表

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

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

随机分布空 null 值
（1）设置 5 个 reduce 个数

hive (default)> 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;

（4）结果：如下图所示

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

hive (default)> 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';
OK
Time taken: 0.368 seconds

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

测试大表直接 JOIN

hive (default)> 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;
OK
b.id	b.t	b.uid	b.keyword	b.url_rank	b.click_num	b.click_url
Time taken: 396.839 seconds

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

hive (default)> 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';
OK
Time taken: 0.537 seconds

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

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

hive (default)> create table bigtable_buck2(
              > 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';

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

（4）设置参数

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

（5）测试

hive (default)> 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;

1.4.3 Group By

以上是关于大数据技术之Hive企业级调优Hive实战的主要内容，如果未能解决你的问题，请参考以下文章

大数据技术之_08_Hive学习_04_压缩和存储（Hive高级）+ 企业级调优（Hive优化）

Hive参数与性能企业级调优

Hive参数与性能企业级调优

Hive参数与性能企业级调优

Hive参数与性能企业级调优

Hive参数与性能企业级调优