Hive调优

Explain查看执行计划

• 在查询语句前加explain

  1explain select * from table;
  2
  3# 查看详细执行计划
  4explain extended select * from table;
  

建表优化

分区表

1# 创建分区表
2create table dept_partition(
3deptno int, dname string, loc string
4)
5partitioned by (day string)
6row format delimited fields terminated by '\t';
7
8# 加载数据到分区表中
9load data local inpath '/opt/module/data/dept_20220221.log' into table dept_partition partition(day='20220221');
10
11# 单分区查询数据
12select * from dept_partition where day='20220221';
13
14# 多分区联合查询
15select * from dept_partition where day='20200401'
16union
17select * from dept_partition where day='20200402'
18
19# 增加单个分区
20alter table dept_partition add partition(day='20220404');
21
22# 同时删除多个分区
23alter table dept_partition drop partition (day='20220404'), partition(day='20220405');
24
25# 查看分区表有多少分区
26show partitions dept_partition;
27
28# 查看分区表结构
29desc formatted dept_partition;

二级分区

1# 创建二级分区表
2create table dept_partition2(
3    deptno int, 
4    dname string, 
5    loc string)
6    partitioned by (day string, hour string)
7    row format delimited fields terminated by '\t';
8
9# 正常加载数据
10load data local inpath '/opt/module/data/dept_20220401.log' into table
11dept_partition2 partition(day='20220401', hour='12');
12
13# 查询分区数据
14select * from dept_partition2 where day='20200401' and hour='12';

动态分区

1# 开启动态分区功能,默认true
2set hive.exec.dynamic.partition=true;
3
4# 设置为非严格模式,默认为严格模式，必须指定至少一个静态分区
5set hive.exec.dynamic.partition.mode=nonstrict;
6
7# 所有MR节点最大可以创建的动态分区数
8set hive.exec.max.dynamic.partitions=1000;
9
10# 每个MR节点上最大可用创建的动态分区数，根据实际情况定
11set hive.exec.max.dynamic.partitions.pernode=365
12
13# 整个MRJob中，最大可以创建的HDFS文件数
14set hive.exec.max.created.files=100000

分桶表

• 分区针对的是数据的存储路径，分桶针对的是数据文件
• 注意事项
  • reduce个数设置为-1,让Job自行决定需要用多少个reduce
  • 从hdfs中load数据到分桶表中，避免本地文件找不到
  • 不要使用本地模式

1# 创建分桶表
2create table stu_buck(id int, name string)
3clustered by(id) 
4into 4 buckets
5row format delimited fields terminated by '\t';
6
7# 查看表结构
8desc formatted stu_buck;
9
10# load方式导入数据到分桶表中
11load data inpath  '/student.txt' into table stu_buck;
12
13# 查询分桶的数据
14select * from stu_buck;

抽样查询

1select * from stu_buck tablesample(bucket 1 out of 4 on id);

HQL语法优化

group by

1# 是否在Map端进行聚合，默认为True
2set hive.map.aggr = true;
3
4# 在Map端进行聚合操作的条目数目
5set hive.groupby.mapaggr.checkinterval = 100000;
6
7# 有数据倾斜的时候进行负载均衡（默认是false）
8set hive.groupby.skewindata = true;

Vectorization

1# 批处理，比单条记录单次获得效率更高
2set hive.vectorized.execution.enabled = true;
3set hive.vectorized.execution.reduce.enabled = true;

多重模式

1insert int t_ptn partition(city=A). select id,name,sex, age from student where city= A;
2insert int t_ptn partition(city=B). select id,name,sex, age from student where city= B;
3
4# 一次读取，多次插入，修改为：
5from student
6insert int t_ptn partition(city=A) select id,name,sex, age where city= A
7insert int t_ptn partition(city=B) select id,name,sex, age where city= B

in/exists语句

1select a.id, a.name from a where a.id in (select b.id from b);
2select a.id, a.name from a where exists (select id from b where a.id = b.id);
3
4# 可以使用Join来改写
5select a.id, a.name from a join b on a.id = b.id;
6
7# 应该转换成 left semi join
8select a.id, a.name from a left semi join b on a.id = b.id;

CBO优化(成本优化器)

1# join的时候，前面的表会被加载到内存中，后面的表进行磁盘扫描
2# 开启CBO，以最小的代价执行最好的代价
3set hive.cbo.enable=true;
4set hive.compute.query.using.stats=true;
5set hive.stats.fetch.column.stats=true;
6set hive.stats.fetch.partition.stats=true;

谓词下推

1# 谓词下推，默认是true
2set hive.optimize.ppd = true; 

MapJoin

1# 设置自动选择MapJoin,默认为true
2set hive.auto.convert.join=true;
3
4# 大表小表的阈值设置（默认25M以下认为是小表）
5set hive.mapjoin.smalltable.filesize=25000000;

大表SMB Join

1# 设置参数
2set hive.optimize.bucketmapjoin = true;
3set hive.optimize.bucketmapjoin.sortedmerge = true;
4set hive.input.format=org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat;

1# 创建分桶表一
2create table bigtable_buck1(
3    id bigint,
4    t bigint,
5    uid string,
6    keyword string,
7    url_rank int,
8    click_num int,
9    click_url string)
10clustered by(id)
11sorted by(id)
12into 6 buckets
13row format delimited fields terminated by '\t';
14load data local inpath '/opt/module/data/bigtable' into table bigtable_buck1;
15
16# 创建分桶表二，分桶数和第一张表的分桶数成倍数关系
17create table bigtable_buck2(
18    id bigint,
19    t bigint,
20    uid string,
21    keyword string,
22    url_rank int,
23    click_num int,
24    click_url string)
25clustered by(id)
26sorted by(id)
27into 6 buckets
28row format delimited fields terminated by '\t';
29load data local inpath '/opt/module/data/bigtable' into table bigtable_buck2;
30
31# 测试
32insert overwrite table jointable
33select b.id, b.t, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url
34from bigtable_buck1 s
35join bigtable_buck2 b
36on b.id = s.id;

数据倾斜

单表数据倾斜

• 使用参数

  1# 是否在Map端进行聚合，默认为True
  2set hive.map.aggr = true;
  3# 在Map端进行聚合操作的条目数目
  4set hive.groupby.mapaggr.checkinterval = 100000;
  5# 有数据倾斜的时候进行负载均衡（默认是false）
  6set hive.groupby.skewindata = true;
  

• 增加Reduce数量（多个key同时导致数据倾斜）

  1# 每个Reduce处理的数据量默认是256MB
  2set hive.exec.reducers.bytes.per.reducer = 256000000
  3# 每个任务最大的reduce数，默认为1009
  4set hive.exec.reducers.max = 1009
  

Join数据倾斜

1# join的键对应的记录条数超过这个值则会进行分拆，值根据具体数据量设置
2set hive.skewjoin.key=100000;
3# 如果是join过程出现倾斜应该设置为true
4set hive.optimize.skewjoin=false;

Job优化

Hive Map优化

• 增加map数

  1# 设置最大切片值为100个字节
  2set mapreduce.input.fileinputformat.split.maxsize=100;
  

• 小文件合并

  1# 在map-only任务结束时合并小文件，默认true
  2set hive.merge.mapfiles = true;
  3# 在map-reduce任务结束时合并小文件，默认false
  4set hive.merge.mapredfiles = true;
  5# 合并文件的大小，默认256M
  6set hive.merge.size.per.task = 268435456;
  7# 当输出文件的平均大小小于该值时，启动一个独立的map-reduce任务进行文件merge
  8set hive.merge.smallfiles.avgsize = 16777216;
  

• Map端聚合

  1# 相当于map端执行combiner
  2set hive.map.aggr=true;
  

• 推测执行

  1# 为拖后腿任务启动一个备份任务，同时运行。谁先运行完，则采用谁的结果
  2set mapred.map.tasks.speculative.execution = true
  

Hive Reduce优化

• 合理设置Reduce数

  1# 每个Reduce处理的数据量默认是256MB
  2set hive.exec.reducers.bytes.per.reducer = 256000000
  3# 每个任务最大的reduce数，默认为1009
  4set hive.exec.reducers.max = 1009
  

• reduce个数并不是越多越好

  • 启动和初始化reduce会消耗时间和资源
  • 有多少个reduce，就会有多少个输出文件，容易造成小文件过多

任务整体优化

• Fetch抓取

  1<property>
  2    <name>hive.fetch.task.conversion</name>
  3    <value>more</value>
  4</property>
  

• 本地模式

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

• 并行执行（数据量很大，sql很长的时候使用）

  1# 打开任务并行执行，默认为false
  2set hive.exec.parallel=true; 
  3# 同一个sql允许最大并行度，默认为8
  4set hive.exec.parallel.thread.number=16; 
  

• JVM重用（小文件过多的时候用）

• 严格模式

  1# 分区表不使用分区过滤
  2# order by没有limit过滤
  3# 笛卡尔积
  4hive.strict.checks.no.partition.filter=true
  

Hive on Spark

1set hive.execution.engine=spark;
2set spark.executor.memory=11.2g;
3set spark.yarn.executor.memoryOverhead=2.8g;
4set spark.executor.cores=4;
5set spark.executor.instances=40;
6set spark.dynamicAllocation.enabled=true;
7set spark.serializer=org.apache.spark.serializer.KryoSerializer;