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= Bin/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 = 1009reduce个数并不是越多越好
- 启动和初始化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;
