大数据4_03_hive调优

10 企业级调优

10.1 Fetch抓取

fetch抓取是指，Hive中对某些情况的查询可以不使用mapreduce计算。

fetch的等级有三个：more（默认）、minimal（老版本）、none

设置为more，在全局查找、字段查找、limit等都不走mapreduce。

hive-default.xml

<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 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)
    </description>
</property>

案例实操：

# 1 将hive.fetch.task.conversion设置成none，然后执行查询语句，都会执行mapreduce程序。
set hive.fetch.task.conversion=none;

# 2 设置成more格式（默认）

10.2 hive的本地模式

大多数的Hadoop Job是需要Hadoop集群提供完整的可扩展性来处理大数据集。不过有的hive的输入数据量是非常小的。在这种情况下，查询触发执行任务消耗的时间可能会比实际job的执行时间要多的多。

• Hive可以通过本地模式在单台机器上处理所有的任务。对于小数据集，执行时间可以明显缩短。

开启本地模式

set hive.exec.mode.local.auto=true;		//开启本地mr

什么时候走本地模式？取决于输入数据量的大小(默认128M)、最大输入文件的个数(默认4个)

set hive.exec.mode.local.auto.inputbytes.max=50000000;		
set hive.exec.mode.local.auto.input.files.max=10;

select * from emp cluster by deptno;
# 不开启本地模式
15 rows selected (29.498 seconds)
# 开启本地模式
15 rows selected (25.555 seconds)

10.3 表的优化

①小表、大表join

多大的表是小表呢？

• 系统默认25M以下是小表。

小表join大表可以减少内存溢出错误的几率，可以使用map join 让小的表先进内存，在map端完成reduce。

• 新版本的hive对小表join大表、大表join小表进行了优化。没有明显的区别。

操作步骤：

# 步骤1 创建大表
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';
# 步骤2 创建小表
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';
# 步骤3 创建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';

# 步骤4 向大表导入数据
load data local inpath '/opt/module/hive/datas/bigtable' into table bigtable;
# 步骤5 向小表导入数据
load data local inpath '/opt/module/hive/datas/smalltable' into table smalltable;

# 步骤6 关闭mapjoin功能（默认开启的）
set hive.auto.convert.join = false;

# 步骤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 smalltable s
join bigtable  b
on b.id = s.id;
# 时间：No rows affected (52.632 seconds)

# 步骤8 执行大表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;
# 时间：No rows affected (52.258 seconds)

②大表join大表

空key过滤

join超时有可能是某些key’对应的数据太多，而相同的key对应的数据都会发送到相同的reducer上，从而导致内存不足。分析异常的key，比如null值，进行提前过滤。

# 步骤1 创建原始表
create table ori(id bigint, t bigint, uid string, keyword string, url_rank int, click_num int, click_url string) row format delimited fields terminated by '\t';
# 步骤2 创建空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 创建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';

# 步骤4 加载数据到原始表，空id表，join后的表
load data local inpath '/opt/module/hive/datas/ori' into table ori;
load data local inpath '/opt/module/hive/datas/nullid' into table nullidtable;

# 步骤5 测试不过滤空id
insert overwrite table jointable select n.* from nullidtable n
left join ori o on n.id = o.id;
# 时间 No rows affected (56.247 seconds)

# 步骤6 测试过滤空id
insert overwrite table jointable select n.* from (select * from nullidtable where id is not null ) n  left join ori o on n.id = o.id;
# 时间 No rows affected (49.802 seconds)

空key转换

对于很多的key为null值，但是该数据可能不是异常数据，所以不能使用空key过滤。为了能让数据均匀的分到不同的reducer上，需要给空key赋一个随机的值。

# 设置5个reduce个数
set mapreduce.job.reduces=5;
# join两张表
insert overwrite table jointable
select n.* from nullidtable n left join ori b on n.id = b.id;
# 时间 No rows affected (68.303 seconds)

数据倾斜

相同key太多了就会使得某个reduce的时间太长。

解决数据倾斜的方式：

• key值得处理
• 不进行reduce操作，操作在map阶段完成。

# 空key转换可以减少数据倾斜
set mapreduce.job.reduces=5;
# join两张表
insert overwrite table jointable
select n.* from nullidtable n full join ori o on 
nvl(n.id,rand()) = o.id;
# 时间 No rows affected (61.956 seconds)

③Map join

map join可以把小表加载到内存在map端join，避免reduce处理。

• 多大的表是小表？

25M以下的认为是小表。

开启map join（默认true）

set hive.auto.convert.join = true;

设置小表的规定阈值(默认25M)

set hive.mapjoin.smalltable.filesize=25000000;

map端的聚合和map join区别？

④Group by

默认情况下，map阶段同一key会分发到一个reduce，当某个key数据太大就会发生数据倾斜。

可以在map端部分聚合，最后在reduce端得到最终结果。

开启map端聚合(默认true)

set hive.map.aggr = true;

设置map端聚合操作的条目数

set hive.groupby.mapaggr.checkinterval = 100000

有数据倾斜的时候进行负载均衡（默认false）

设置了负载均衡true会有两个mr job，第一个mr job中map的结果会随机分到reduce中，目的是相同的group by key 有可能会分到不同的reduce中，从而达到负载均衡的目的。第二个mr job在根据预处理的结果按照group by key分配到reduce中。

• 大数据效果明显。

set hive.groupby.skewindata = true

⑤关闭笛卡尔积

⑥行列过滤

行处理：少用select(*)

列处理：两个表join的时候，先条件过滤，在关联两个表。

⑦动态分区、分桶、分区

10.4 MR优化

①设置map个数

• map的个数=切片数

• map是越多越好吗？并不是！（减小map数）

  如果一个文件有很多小文件，每一个小文件会被当做一个块，用一个map来完成，map的任务启动时间和初始化时间远远大于逻辑处理时间，就会造成资源的浪费。

• 那么是不是输入文件都是128M就高枕无忧了？并不是（增大map数）

  如果有一个127M的文件，里面的每一行只有一个单词，那么map处理每行就特别耗时。

减少map数

小文件合并，Hive提供CombineHiveInputformat，对小文件合并，默认开启

set hive.input.format= org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

增大map数

computeSliteSize(Math.max(minSize,Math.min(maxSize,blocksize)))=blocksize=128M公式，调整maxSize最大值。让maxSize最大值低于blocksize就可以增加map的个数。

设置最大切片值（默认100字节）

set mapreduce.input.fileinputformat.split.maxsize=100;

②设置reduce个数

• reduce的个数可以自己设置，默认为-1；

• 如果输入一个1G的文件，默认情况下会有几个reduce？

  每个reduce处理的数据量默认是256M；那么1G/256M≈4个

• 每个任务的最大reduce个数=1009

• 真实的生产环境reduce的个数上千个。

方式1：

set mapreduce.job.reduces=15;

方式2：

# 每个reduce默认处理的数据量是256M
set hive.exec.reducers.bytes.per.reducer=256000000;
# 每个任务的最大reduce数，默认是1009
set hive.exec.reducers.max=1009;
# 计算reduce数的公式
N = min(1009, 总的输入量/256000000)

10.5 并行执行

并行执行就是map阶段没有完全完成的情况下，map处理的部分数据会传给reduce，reduce先聚合这一部分，最后在聚合所有的。

未开启并行执行：

开启并行执行：

开启并行阶段取决于资源量，同时进行map和reduce比较耗资源。

set hive.exec.parallel=true;		//打开任务并行度，默认false
set hive.exec.parallel.thread.number=16;		//同一个sql允许最大并行度，默认8

10.6 严格模式

Hive的严格模式具有以下功能（默认nonstrict），生产环境（开启strict）

<property>
    <name>hive.mapred.mode</name>
    <value>strict</value>
    <description>
      The mode in which the Hive operations are being performed. 
      In strict mode, some risky queries are not allowed to run. They include:
        Cartesian Product.		//关闭笛卡尔积
        No partition being picked up for a query.		//分区表
        Comparing bigints and strings.			//比较
        Comparing bigints and doubles.			//比较
        Orderby without limit.			//使用order by必须还是不必须加limit
	</description>
</property>

动态分区默认是strict严格模式（是必须指定一个分区），设置成nonstrict非严格模式（不需要必须指定一个分区）

10.7 JVM重用

JVM重用是Hadoop调优参数的内容，对hive的性能具有非常大的影响，特别是对小文件场景或task特别多的场景，这类场景大多数执行时间都很短。

JVM重用可以使得JVM实例在同一个job中重新使用N次。N的值在Hadoop的mapred-site.xml文件中进行配置。通常在10-20之间。

<property>
  <name>mapreduce.job.jvm.numtasks</name>
  <value>10</value>
  <description>How many tasks to run per jvm. If set to -1, there is
  no limit. 
  </description>
</property>

10.8 压缩

10.9 执行计划Explain

explain select * from emp;

explain select deptno, avg(sal) avg_sal from emp group by deptno;

# 查看详细执行计划
explain extended select * from emp;