1、基础优化
(1) 列裁剪和分区裁剪
最基本的操作。所谓列裁剪就是在查询时只读取需要的列，分区裁剪就是只读取需要的分区。
SELECT
uid,
event_type,
record_data
FROM
calendar_record_log
WHERE
pt_date >= 20190201 AND pt_date <= 20190224 AND STATUS = 0;
当列很多或者数据量很大时，如果select * 或者不指定分区，全列扫描和全表扫描效率都很低。
Hive中与列裁剪优化相关的配置项是hive.optimize.cp，与分区裁剪优化相关的则是hive.optimize.pruner，默认都是true。
在HiveSQL解析阶段对应的则是ColumnPruner逻辑优化器。
(2) 谓词下推
在关系型数据库如MySQL中，也有谓词下推（Predicate Pushdown，PPD）的概念。它就是将SQL语句中的where谓词逻辑都尽可能提前执行，减少下游处理的数据量。
SELECT
a.uid,
a.event_type,
b.topic_id,
b.title
FROM
calendar_record_log a
LEFT OUTER JOIN
( SELECT uid, topic_id, title FROM forum_topic WHERE pt_date = 20190224 AND length( content ) >= 100 ) b
ON a.uid = b.uid
WHERE a.pt_date = 20190224 AND STATUS = 0;
对forum_topic做过滤的where语句写在子查询内部，而不是外部。Hive中有谓词下推优化的配置项hive.optimize.ppd，默认值true，与它对应的逻辑优化器是PredicatePushDown。
该优化器就是将OperatorTree中的FilterOperator向上提。
(3) sort by代替order by
Hive中的全局排序，使用Order By来进行。在Hive中使用全局排序时，需要注意，Hive会将所有数据交给一个Reduce任务计算，实现查询结果的全局排序。
所以如果数据量很大，只有一个Reduce会耗费大量时间。
如果使用sort by，那么还是会视情况启动多个reducer进行排序，并且保证每个reducer内局部有序。为了控制map端数据分配到reducer的key，往往还要配合distribute by一同使用。
如果不加distribute by的话，map端数据就会随机分配到reducer。
举个例子，假如要以UID为key，以上传时间倒序、记录类型倒序输出记录数据：
select
uid,upload_time,event_type,record_data
from
calendar_record_log
where pt_date >= 20190201 and pt_date <= 20190224
distribute by uid
sort by upload_time desc,event_type desc;
全局排序优化:
(1) 一般而言，对全量数据进行全局排序的场景很少，一般只需要保证查询结果最终有序即可，这时可以先使用子查询得到一个小的结果集，然后再进行排序。
select * from
(select id,count(1) cnt from <table_name> where id !='0' group by user_id) a
order by a.cnt;
(2) 如果是取TOP N的情况，则可以使用子查询，在每个Reduce中进行排序后，各自取得前N个数据，然后再对结果集进行全局排序，最终取得结果。
select a.id,a.name
from
( select id,name from <table_name> distribute by length(name) sort by length(name) desc limit 10 ) a
order by length(a.user_name) desc limit 10;
(4) group by代替distinct
当要统计某一列的去重数时，如果数据量很大，count(distinct)就会非常慢，原因与order by类似，count(distinct)逻辑只会有很少的reducer来处理。
这时可以用group by来改写：
select
count(1) as cnt
from
( select uid from calendar_record_log where pt_date >= 20190101 group by uid ) t;
但是这样写会启动两个MR job（单纯distinct只会启动一个），所以要确保数据量大到启动job的overhead远小于计算耗时，才考虑这种方法。
当数据集很小或者key的倾斜比较明显时，group by还可能会比distinct慢。
那么如何用group by方式同时统计多个列？：
select
t.a,sum(t.b),count(t.c),count(t.d)
from (
select a,b,null c,null d from some_table
union all
select a,0 b,c,null d from some_table group by a,c
union all
select a,0 b,null c,d from some_table group by a,d
) t;
2、group by优化
(1) map端预聚合，默认开启
group by时，如果先起一个combiner在map端做部分预聚合，可以有效减少shuffle数据量。预聚合的配置项是hive.map.aggr，默认值true，对应的优化器为GroupByOptimizer。
通过hive.groupby.mapaggr.checkinterval参数也可以设置map端预聚合的行数阈值，超过该值就会分拆job，默认值100000。
(2) 倾斜均衡配置项，需要手动开启（双重聚合）
group by时如果某些key对应的数据量过大，就会发生数据倾斜。Hive自带了一个均衡数据倾斜的配置项hive.groupby.skewindata，默认值false。
实现方法是在group by时启动两个MR job：
第一个job会将map端数据随机输入到reducer，每个reducer做部分聚合，相同的key就会分布在不同的reducer中。
第二个job再将前面预处理过的数据按key聚合并输出结果，这样就起到了均衡的效果。
配置项毕竟是死的，单纯靠它有时不能根本上解决问题，因此还是建议自行了解数据倾斜的细节，并优化查询语句。
3、join优化
(1) build table（小表）前置
在最常见的hash join方法中，一般总有一张相对小的表和一张相对大的表，小表叫build table，大表叫probe table。
Hive在解析带join的SQL语句时，会默认将最后一个表作为probe table，将前面的表作为build table并试图将它们读进内存。
如果表顺序写反，probe table在前面，引发OOM的风险就高了。
在维度建模数据仓库中，事实表就是probe table，维度表就是build table。假设现在要将日历记录事实表和记录项编码维度表来join：
select
a.event_type,a.event_code,a.event_desc,b.upload_time
from
calendar_event_code a
inner join
( select event_type,upload_time from calendar_record_log where pt_date = 20190225 ) b on a.event_type = b.event_type;
(2) map端join
map join特别适合大小表join的情况。Hive会将build table和probe table在map端直接完成join过程，消灭了reduce，效率很高。
select
a.event_type,b.upload_time
from
calendar_event_code a
inner join
( select event_type,upload_time from calendar_record_log where pt_date = 20190225 ) b
on a.event_type < b.event_type;
map join的配置项是hive.auto.convert.join，默认值true，对应逻辑优化器是MapJoinProcessor。
还有一些参数用来控制map join的行为，比如hive.mapjoin.smalltable.filesize，当build table大小小于该值就会启用map join，默认值25000000（25MB）。
还有hive.mapjoin.cache.numrows，表示缓存build table的多少行数据到内存，默认值25000。
分桶表map join
map join对分桶表还有特别的优化。由于分桶表是基于一列进行hash存储的，因此非常适合抽样（按桶或按块抽样）。
它对应的配置项是hive.optimize.bucketmapjoin，优化器是BucketMapJoinOptimizer。
(3) join时候数据倾斜的倾斜均衡配置项（转换为map join）
这个配置与上面group by的倾斜均衡配置项异曲同工，通过hive.optimize.skewjoin来配置，默认false。
开启此参数后，解决join时候产生数据倾斜的原理：
如果开启了，在join过程中Hive会将计数超过阈值hive.skewjoin.key（默认100000）的倾斜key对应的行临时写进文件中，然后再启动另一个job做map join生成结果。
通过hive.skewjoin.mapjoin.map.tasks参数还可以控制第二个job的mapper数量，默认10000。
再重复一遍，通过自带的配置项经常不能解决数据倾斜问题。
4、SQL层面处理数据倾斜问题
(1) null 或 无意义值
这种情况很常见，比如当事实表是日志类数据时，往往会有一些项没有记录到，我们视情况会将它置为null，或者空字符串、-1等。
表中有大量的null值，如果表之间进行join操作，就会有shuffle产生，这样所有的null值都会被分配到一个reduce中，必然产生数据倾斜。
解决方案：
第一种：可以直接不让null值参与join操作，即不让null值有shuffle阶段
SELECT
*
FROM
log a
JOIN
users b
ON a.user_id IS NOT NULL AND a.user_id = b.user_id
UNION ALL
SELECT
*
FROM log a
WHERE a.user_id IS NULL;
第二种：因为null值参与shuffle时的hash结果是一样的，那么我们可以给null值随机赋值，这样它们的hash结果就不一样，就会进到不同的reduce中：
SELECT
*
FROM
log a
LEFT JOIN
users b
ON CASE WHEN a.user_id IS NULL THEN concat('hive_', rand()) ELSE a.user_id END = b.user_id;
(2) 单独处理倾斜key
这其实是上面处理空值方法的拓展，不过倾斜的key变成了有意义的。一般来讲倾斜的key都很少，我们可以将它们抽样出来，对应的行单独存入临时表中，
然后打上一个较小的随机数前缀（比如0~9），最后再进行聚合。SQL语句与上面的相仿，不再赘述。
(3) 不同数据类型引发的数据倾斜
对于两个表join，表a中需要join的字段key为int，表b中key字段既有string类型也有int类型。
当按照key进行两个表的join操作时，默认的Hash操作会按int型的id来进行分配，这样所有的string类型都被分配成同一个id，结果就是所有的string类型的字段进入到一个reduce中，
引发数据倾斜。
解决方案：
如果key字段既有string类型也有int类型，默认的hash就都会按int类型来分配，那我们直接把int类型都转为string就好了，这样key字段都为string，hash时就按照string类型分配了：
SELECT
*
FROM users a
LEFT JOIN logs b ON a.usr_id = CAST(b.user_id AS string);
(4) build table过大
有时，build table会大到无法直接使用map join的地步，比如全量用户维度表，而使用普通join又有数据分布不均的问题。
这时就要充分利用probe table的限制条件，削减build table的数据量，再使用map join解决。代价就是需要进行两次join。
select
a.uid,a.event_type,b.status,b.extra_info
from
calendar_record_log a
left outer join
(
select
t.uid,t.status,t.extra_info
from
(select distinct uid from calendar_record_log where pt_date = 20190228) s
inner join
user_info t on s.uid = t.uid
) b
on a.uid = b.uid
where a.pt_date = 20190228;
5、MR 优化
(1) 调整mapper数
可以参考：https://blog.csdn.net/qq_44665283/article/details/124975349
(2) 调整reducer数
reducer数量的确定方法比mapper简单得多。使用参数mapred.reduce.tasks可以直接设定reducer数量，不像mapper一样是期望值。
但如果不设这个参数的话，Hive就会自行推测，逻辑如下：
参数hive.exec.reducers.bytes.per.reducer用来设定每个reducer能够处理的最大数据量，256M（1.2版本之后）。
参数hive.exec.reducers.max用来设定每个job的最大reducer数量，默认值1009（1.2版本之后）。
得出reducer数：
reducer_num = MIN(total_input_size / reducers.bytes.per.reducer, reducers.max)。
reducer数量与输出文件的数量相关。
如果reducer数太多，会产生大量小文件，对HDFS造成压力。
如果reducer数太少，每个reducer要处理很多数据，容易拖慢运行时间或者造成OOM。
（3）合并小文件
1、输入阶段合并
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; #默认
set hive.input.format=org.apache.hadoop.mapred.TextInputFormat
SET hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
2、输出阶段合并
直接将hive.merge.mapfiles和hive.merge.mapredfiles都设为true即可，前者表示将map-only任务的输出合并，后者表示将map-reduce任务的输出合并。
另外，hive.merge.size.per.task可以指定每个task输出后合并文件大小的期望值，hive.merge.size.smallfiles.avgsize可以指定所有输出文件大小的均值阈值，默认值都是1GB。
如果平均大小不足的话，就会另外启动一个任务来进行合并。
3、启用压缩
压缩job的中间结果数据和输出数据，可以用少量CPU时间节省很多空间。压缩方式一般选择Snappy，效率最高。
要启用中间压缩，需要设定hive.exec.compress.intermediate为true，同时指定压缩方式hive.intermediate.compression.codec为org.apache.hadoop.io.compress.SnappyCodec。
另外，参数hive.intermediate.compression.type可以选择对块（BLOCK）还是记录（RECORD）压缩，BLOCK的压缩率比较高。
输出压缩的配置基本相同，打开hive.exec.compress.output即可。
4、JVM重用
在MR job中，默认是每执行一个task就启动一个JVM。如果task非常小而碎，那么JVM启动和关闭的耗时就会很长。可以通过调节参数mapred.job.reuse.jvm.num.tasks来重用。
例如将这个参数设成5，那么就代表同一个MR job中顺序执行的5个task可以重复使用一个JVM，减少启动和关闭的开销。但它对不同MR job中的task无效。
6、严格模式
所谓严格模式，就是强制不允许用户执行3种有风险的HiveSQL语句，一旦执行会直接失败。这3种语句是：
- 查询分区表时不限定分区列的语句；
- 两表join产生了笛卡尔积的语句；
- 用order by来排序但没有指定limit的语句。
要开启严格模式，需要将参数hive.mapred.mode设为strict。
7、并行执行与本地模式
并行执行
Hive中互相没有依赖关系的job间是可以并行执行的，最典型的就是多个子查询union all。
在集群资源相对充足的情况下，可以开启并行执行，即将参数hive.exec.parallel设为true。
另外hive.exec.parallel.thread.number可以设定并行执行的线程数，默认为8，一般都够用。
本地模式
Hive也可以不将任务提交到集群进行运算，而是直接在一台节点上处理。因为消除了提交到集群的overhead，所以比较适合数据量很小，且逻辑不复杂的任务。
设置hive.exec.mode.local.auto为true可以开启本地模式。
但任务的输入数据总量必须小于hive.exec.mode.local.auto.inputbytes.max（默认值128MB），且mapper数必须小于hive.exec.mode.local.auto.tasks.max（默认值4），
reducer数必须为0或1，才会真正用本地模式执行。
|