Hive调优

Posted 2022-02-17 李煊的个人博客

Explain查看执行计划

• 在查询语句前加explain

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

建表优化

分区表

# 创建分区表
create table dept_partition(
deptno int, dname string, loc string
)
partitioned by (day string)
row format delimited fields terminated by \'\\t\';

# 加载数据到分区表中
load data local inpath \'/opt/module/data/dept_20220221.log\' into table dept_partition partition(day=\'20220221\');

# 单分区查询数据
select * from dept_partition where day=\'20220221\';

# 多分区联合查询
select * from dept_partition where day=\'20200401\'
union
select * from dept_partition where day=\'20200402\'

# 增加单个分区
alter table dept_partition add partition(day=\'20220404\');

# 同时删除多个分区
alter table dept_partition drop partition (day=\'20220404\'), partition(day=\'20220405\');

# 查看分区表有多少分区
show partitions dept_partition;

# 查看分区表结构
desc formatted dept_partition;

二级分区

# 创建二级分区表
create table dept_partition2(
    deptno int, 
    dname string, 
    loc string)
    partitioned by (day string, hour string)
    row format delimited fields terminated by \'\\t\';

# 正常加载数据
load data local inpath \'/opt/module/data/dept_20220401.log\' into table
dept_partition2 partition(day=\'20220401\', hour=\'12\');

# 查询分区数据
select * from dept_partition2 where day=\'20200401\' and hour=\'12\';

动态分区

# 开启动态分区功能,默认true
set hive.exec.dynamic.partition=true;

# 设置为非严格模式,默认为严格模式，必须指定至少一个静态分区
set hive.exec.dynamic.partition.mode=nonstrict;

# 所有MR节点最大可以创建的动态分区数
set hive.exec.max.dynamic.partitions=1000;

# 每个MR节点上最大可用创建的动态分区数，根据实际情况定
set hive.exec.max.dynamic.partitions.pernode=365

# 整个MRJob中，最大可以创建的HDFS文件数
set hive.exec.max.created.files=100000

分桶表

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

# 创建分桶表
create table stu_buck(id int, name string)
clustered by(id) 
into 4 buckets
row format delimited fields terminated by \'\\t\';

# 查看表结构
desc formatted stu_buck;

# load方式导入数据到分桶表中
load data inpath  \'/student.txt\' into table stu_buck;

# 查询分桶的数据
select * from stu_buck;

抽样查询

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

HQL语法优化

group by

# 是否在Map端进行聚合，默认为True
set hive.map.aggr = true;

# 在Map端进行聚合操作的条目数目
set hive.groupby.mapaggr.checkinterval = 100000;

# 有数据倾斜的时候进行负载均衡（默认是false）
set hive.groupby.skewindata = true;

Vectorization

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

多重模式

insert int t_ptn partition(city=A). select id,name,sex, age from student where city= A;
insert int t_ptn partition(city=B). select id,name,sex, age from student where city= B;

# 一次读取，多次插入，修改为：
from student
insert int t_ptn partition(city=A) select id,name,sex, age where city= A
insert int t_ptn partition(city=B) select id,name,sex, age where city= B

in/exists语句

select a.id, a.name from a where a.id in (select b.id from b);
select a.id, a.name from a where exists (select id from b where a.id = b.id);

# 可以使用Join来改写
select a.id, a.name from a join b on a.id = b.id;

# 应该转换成 left semi join
select a.id, a.name from a left semi join b on a.id = b.id;

CBO优化(成本优化器)

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

谓词下推

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

MapJoin

# 设置自动选择MapJoin,默认为true
set hive.auto.convert.join=true;

# 大表小表的阈值设置（默认25M以下认为是小表）
set hive.mapjoin.smalltable.filesize=25000000;

大表SMB Join

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

# 创建分桶表一
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;

# 创建分桶表二，分桶数和第一张表的分桶数成倍数关系
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\';
load data local inpath \'/opt/module/data/bigtable\' into table bigtable_buck2;

# 测试
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;

数据倾斜

单表数据倾斜

• 使用参数

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

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

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

Join数据倾斜

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

Job优化

Hive Map优化

• 增加map数

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

• 小文件合并

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

• Map端聚合

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

• 推测执行

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

Hive Reduce优化

• 合理设置Reduce数

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

• reduce个数并不是越多越好

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

任务整体优化

• Fetch抓取

  <property>
      <name>hive.fetch.task.conversion</name>
      <value>more</value>
  </property>
  

• 本地模式

  # 开启本地模式
  set hive.exec.mode.local.auto=true; 
  # 设置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;
  

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

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

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

• 严格模式

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

Hive on Spark

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