Hive优化以及相关参数设置

1.表层面设计优化

1.1 表分区

分区表实际上就是对应一个 HDFS 文件系统上的独立的文件夹，该文件夹下是该分区所有的数据文件。Hive 中的分区就是分目录，把一个大的数据集根据业务需要分割成小的数据集。在查询时通过 WHERE 子句中的表达式选择查询所需要的指定的分区，这样的查询效率会提高很多。

使用场景：在进行模型设计的时候如果考虑到表数据量很大，则需要对该表进行分区操作，在实际的项目中，分区一般都是按照日期进行分区

如果现在要向一个分区表中加载数据，则选择使用静态分区，一般增量抽取到的数据进行加载的时候都是用的是静态分区。
如果获取到的数据要按照某一列的值保存到多个分区中，则需要使用动态分区

1.2 表分桶

分区提供一个隔离数据和优化查询的便利方式。不过，并非所有的数据集都可形成合理的分区。对于一张表或者分区，Hive 可以进一步组织成桶，也就是更为细粒度的数据范围划分。

分桶将整个数据内容安装某列属性值得hash值进行区分，如按照name属性分为3个桶，就是对name属性值的hash值对3取摸，按照取模结果对数据分桶。如取模结果为0的数据记录存放到一个文件，取模为1的数据存放到一个文件，取模为2的数据存放到一个文件。分桶是将数据集分解成更容易管理的若干部分的另一个技术。分区针对的是数据的存储路径；分桶针对的是数据文件。

使用场景：1.如果表的数据量比较大，在加载数据的时候想要启用多个reduce则可以使用分桶表。
       2.如果表中的数据量比较大，并且经常按照某一列进行表关联操作，则建议使用分桶表，提高关联效率。
               3.如果分桶表之后还要提高关联效率，则可以在分桶表上对每一桶的数据进行排序。
               建表的时候分桶排序：
                   create table 表名(
                   .......
                   )clustered by(分桶列) sorted by (排序列) into n buckets
                   ....
               4.如果要经常对表中的数据进行抽样查询，则也可以使用分桶表。

1.3 文件存储格式

文件存储格式	缺点	优点	加载数据方式
textfile 不支持切片压缩	磁盘消耗大	直接将本地的文件加载的hdfs中,加载数据的速度最高	1.-put 2.load data 3.insert into
sequencefile 二进制文件,可压缩可切片	读写速度慢	支持切片2.可以进行数据的压缩	insert into
rcfile 文件的存储方式，支持切片，支持压缩			insert into
orc 压缩和文件处理的效率都比rcfile高很多		高效的数据压缩高效的数据处理的速度	insert into
parquet 一种文件存储格式	不支持数据的修改操作	高效的数据压缩高效的数据处理的速度	insert into

TextFile

默认格式，存储方式为行存储，数据不做压缩，磁盘开销大，数据解析开销大。
可结合Gzip、Bzip2使用(系统自动检查，执行查询时自动解压)，但使用这种方式，压缩后的文件不支持split，Hive不会对数据进行切分，从而无法对数据进行并行操作。
并且在反序列化过程中，必须逐个字符判断是不是分隔符和行结束符，因此反序列化开销会比SequenceFile高几十倍。

SequenceFile

SequenceFile是Hadoop API提供的一种二进制文件支持，以key-value的形式序列化到文件中,存储方式为行存储，其具有使用方便、可分割、可压缩的特点。
压缩数据文件可以节省磁盘空间，但Hadoop中有些原生压缩文件的缺点之一就是不支持分割。支持分割的文件可以并行的有多个mapper程序处理大数据文件，大多数文件不支持可分割是因为这些文件只能从头开始读。Sequence File是可分割的文件格式，支持Hadoop的block级压缩。
SequenceFile支持三种压缩选择：NONE，RECORD，BLOCK。RECORD是默认选项,Record压缩率低,通常BLOCK会带来较RECORD更好的压缩性能,一般建议使用BLOCK压缩。
优势是文件和hadoop api中的MapFile是相互兼容的。

RCFile

存储方式：数据按行分块，每块按列存储。结合了行存储和列存储的优点：
RCFile 保证同一行的数据位于同一节点，因此元组重构的开销很低；
像列存储一样，RCFile 能够利用列维度的数据压缩，并且能跳过不必要的列读取；
数据追加：RCFile不支持任意方式的数据写操作，仅提供一种追加接口，这是因为底层的 HDFS当前仅仅支持数据追加写文件尾部。

ORCFile（Optimized Row Columnar）

存储方式：数据按行分块每块按照列存储。
压缩快、快速列存取，提高Hive的读取、写入和处理数据的性能。
效率比rcfile高，是rcfile的优化版本。
能够兼容多种计算引擎

Parquet

是一种列式数据存储格式，可以兼容多种计算引擎，如MapRedcue 和Spark等，对多层嵌套的数据结构提供了良好的性能支持，是目前Hive 生产环境中数据存储的主流选择之一。

建议：在实际的项目中，如果贴源层加载一些本地的文件，则需要使用textfile，数据仓库中间的一些层建议使用orc或者parquet，如果项目只是用hive的计算框架则选择orc，如果项目中用到了多种计算框架则建议使用parquet,最上层的应用层一般会将数据导出到共享层，并且数据量不大所以直接使用textfile。

1.4 压缩格式

【输出结果来设置压缩格式】

--SQL语句的最终输出结果是否压缩：

set hive.exec.compress.output=true;

--输出结果的压缩格式（以snappy为例）：

开启mapreduce最终输出数据压缩

set mapreduce.output.fileoutputformat.compress=true;

set mapreduce.output.fileoutputformat.compress.codec

=org.apache.hadoop.io.compress.SnappyCodec;

【保存文件的时候设置压缩格式】

---建表语句设置压缩格式:

create table table_name(
...
)stored as orc
tblproperties ("orc.compress"="snappy");

【对单个MR的中间结果进行压缩】
单个MR的中间结果指的是Mapper输出的数据，压缩该数据库降低Shuffle阶段的IO压力，配置以下参数：

--开启MapReduce中间数据压缩功能
set mapreduce.map.output.compress=true;

--设置MapReduce中间数据的压缩方式（以snappy为例）
set mapreduce.map.output.compress.codec

=org.apache.hadoop.io.compress.SnappyCodec;

【对单条SQL语句的中间结果进行压缩】
   一个SQL数据可能通过MR进行计算，单条SQL语句的中间结果指两个MR之间的临时数据，配置以下参数：
   --是否对两个MR之间的临时数据进行压缩
   set hive.exec.compress.intermediate=true;
   --设置两个MR之间的压缩格式（以snappy为例）
   set hive.intermediate.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;

压缩格式	算法	文件扩展名	是否可切分
Gzip	DEFLATE	.gz	否
bzip2	bzip2	.bz2	是
LZO	LZO	.lzo	是
Snappy	Snappy	.snappy	否

为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器，如下表所示：

压缩格式	对应的编码/解码器
gzip	org.apache.hadoop.io.compress.GzipCodec
bzip2	org.apache.hadoop.io.compress.BZip2Codec
LZO	com.hadoop.compression.lzo.LzopCodec
Snappy	org.apache.hadoop.io.compress.SnappyCodec

压缩性能的比较：

压缩算法	原始文件大小	压缩文件大小	压缩速度	解压速度
gzip	8.3GB	1.8GB	17.5MB/s	58MB/s
bzip2	8.3GB	1.1GB	2.4MB/s	9.5MB/s
LZO	8.3GB	2.9GB	49.3MB/s	74.6MB/s

压缩方式选择时重点考虑：解压缩速度、压缩率、压缩后是否可以支持切片

压缩格式	是否支持切片	解压缩速度	压缩率
snappy	no	最快	很差
lzo	yes	很快	很高
bzip2	yes	最慢	最高
gzip	no	一般	很高

使用场景：

2.2.1 Gzip压缩
优点：压缩率比较高，而且压缩/解压速度也比较快；Hadoop本身支持，在应用中处理Gzip格式的文件就和直接处理文本一样；大部分Linux系统都自带Gzip命令，使用方便。

缺点：不支持Split。

应用场景：当每个文件压缩之后在130M以内的（1个块大小内），都可以考虑用Gzip压缩格式。例如说一天或者一个小时的日志压缩成一个Gzip文件。

2.2.2 Bzip2压缩
优点：支持Split；具有很高的压缩率，比Gzip压缩率都高；Hadoop本身自带，使用方便。

缺点：压缩/解压速度慢。

应用场景：在数据仓库中处理数据的时候一般很少选择，但是它可以使用在一些对文件进行归档保存

2.2.3 Lzo压缩
优点：压缩/解压速度也比较快，合理的压缩率；支持Split，是Hadoop中最流行的压缩格式；可以在Linux系统下安装lzop命令，使用方便。

缺点：压缩率比Gzip要低一些；Hadoop本身不支持，需要安装；在应用中对Lzo格式的文件需要做一些特殊处理（为了支持Split需要建索引，还需要指定InputFormat为Lzo格式）。

应用场景：一个很大的文本文件，压缩之后还大于200M以上的可以考虑，而且单个文件越大，Lzo优点越越明显。

2.2.4 Snappy压缩
优点：高速压缩速度和合理的压缩率。

缺点：不支持Split；压缩率比Gzip要低；Hadoop本身不支持，需要安装。

应用场景：一般在中间层文件存储的时候会使用，或者是mapreduce中中间临时文件的压缩可以使用，因为他有高效的压缩解压速度，所以一般会配合orc或者parquet一起使用。

压缩参数配置

设置map后输出压缩
1．开启hive中间传输数据压缩功能
set hive.exec.compress.intermediate=true;2．开启mapreduce中map输出压缩功能
set mapreduce.map.output.compress=true;3．设置mapreduce中map输出数据的压缩方式
set mapreduce.map.outout.compress.codec=
org.apache.hadoop.io.compress.SnappyCodec
org.apache.hadoop.io.compress.GzipCodec 
org.apache.hadoop.io.compress.BZip2Codec 
org.apache.hadoop.io.compress.Lz4Codec

设置reduce后输出压缩
1．开启hive最终输出数据压缩功能
set hive.exec.compress.output=true;2．开启mapreduce最终输出数据压缩
set mapreduce.output.fileoutputformat.compress=true;3．设置mapreduce最终数据输出压缩方式
set mapreduce.output.fileoutputformat.compress.codec =
org.apache.hadoop.io.compress.SnappyCodec
org.apache.hadoop.io.compress.GzipCodec 
org.apache.hadoop.io.compress.BZip2Codec 
org.apache.hadoop.io.compress.Lz4Codec4．设置mapreduce最终数据输出压缩为块压缩
set mapreduce.output.fileoutputformat.compress.type=BLOCK;

2.语法和参数的层面

2.1.列裁剪

列裁剪（只查询需要的字段，千万不要直接写 select * from）列裁剪就是在查询时只读取需要的列。当列很多或者数据量很大时，如果select所有的列或者不指定分区，导致的全列扫描和全表扫描效率都很低。

设置列裁剪的参数：set hive.optimize.cp=true

2.2 分区裁剪

在查询数据的时候只选择所需要的分区进行过滤，减少数据的读入量，提高查询效率。
设置分区裁剪的参数：set hive.optimize.pruner=true --开启分区裁剪（默认开启）

2.3.group by优化

1.开启map端聚合
很多操作不一定要在reduce中进行，可以在map端完成聚合操作，然后reduce中直接输出结果即可。

设置开启map聚合参数：set hive.map.aggr=true; ---开启map端聚合操作
设置map端聚合的条数：set hive.groupby.mapaggr.checkinterval=100000

2.解决数据倾斜问题
默认情况下，在进行分组聚合的时候，相同的键的数据会进入到同一个reduce中进行处理，如果分组的时候某一个值有大量的重复的数据，则会导致某一个reduce任务量会很大，从而导致数据倾斜。
解决办法：开启负载均衡 set hive.groupby.skewindata=true; --默认为false
开启负载均衡之后，系统在分组的时候会启动两个mapreduce程序
       第一个mapreduce程序会先在map端将数据按照key平均分配，可能会出现相同的键的数据进入不同的reduce中，然后在reduce中进行初步聚合操作
       然后将第一个mapreduce中reduce聚合的结果交给第二个mapreduce进行处理，然后在map端将相同的键分配到同一个reduce进行最终的聚合操作，最后输出结果。

3.去重优化：
去重方式：distinct和group by都可以
在数据量比较大的情况下建议使用group by来代替distinct。
4.排序优化：
   order by:全局排序，不管数据量多大，都会启动一个reduce进行数据的处理，所以效率比较低。
   sort by:局部排序，会将每一个reduce中的数据进行排序，不能保证全局的数据是有序的，一般配合distribute by一起使用。
   cluster by:和sort by是一样的，但是不能进行降序排序。
   优化建议：1.尽量避免全局排序，不要一开始就对表中的数据进行全局排序，如果要进行排序，最好将排序的操作放到最后面。
                        2.如果要查询表中的前n条数据，则建议使用sort by配合limit一起使用。
   5.hive中为了提高查询效率，尽量每次获取条的条数的时候不要去查询文件，可以直接从元数据中获取条的条数信息。
   参数设置：
       set hive.compute.query.using.stats=true; --在查询表的条数的时候直接从元数据中获取（默认开启）
6.表join优化：
原则：将大表转换成小表进行关联
表的关联方式：
1.map side join:在map端完成关联操作，不需要启动reduce程序。
       使用场景：关联的表中有一个小表则可以使用该种方式（如果有小表进行关联系统默认会使用map side join）

       在hive中默认的小表为小于25M的表，当然可以通过参数来设置小表默认大小：
       set hive.mapjoin.smalltable.filesize=25000000

       设置让系统自动识别小表：
       set hive.auto.convert.join=true --系统默认识别小表

       select /*+mapjoin(小表名)*/ * from 表 a join 表 b on ......
       2.bucket map join:分桶关联
       使用场景：大表和大表关联的时候可以使用，最好关联的表按照关联的列进行分桶

       --设置开启分桶关联：
       set hive.optimize.bucketmapjoin=true; ----默认false

       3. sort merge bucket map join:
       在bucket map join的基础上将每一桶的数据先进行排序，然后再进行关联操作，可以大大的减少表关联次数，提高查询效率。

       --设置开启排序分桶关联：
       set hive.optimize.bucketmapjoin.sortedmerge=true; ---默认关闭false
       注意：如果大表和大表关联的时候，表有分桶并且也有排序，则关联效率最高的。

8.谓词下推：PPD
主要思想就是尽量将过滤条件放在map端执行，这样就可以减少后续的数据的数据量，从而降低了数据传输的IO消耗，提高查询效率。
设置开启谓词下推的参数：
set hive.optimize.ppd=true;       ---默认就是开启

       注意：如果要准确的来测试谓词下推的情况，则必须关闭CBO优化器
       --关闭CBO优化器的参数
       set hive.cbo.enable=false; --默认true开启

谓词：代表sql语句中的过滤条件
主要测试谓词写在on后面和写在where后面的谓词是否会下推。
表关联的时候表主要分为两种：
1.保留表：在进行表关联的时候，如果要保留该表的全部数据则叫做保留表，例如：左连接的左边
2.空行表：   在进行表关联的时候，如果匹配不到的数据使用空行代替的表，叫做空行表，例如：左连接的右边


       内连接：如果两张表进行的是内连接，则不管CBO优化器是否打开，也不管条件在on后面还是where后面，都可以实现谓词下推。
       左外|右外连接：
       情况1：过滤条件写在where后面，并且过滤的字段属于保留表，可以进行谓词下推
       情况2：过滤条件写在where后面，但是过滤的字段属于空行表，不会进行谓词下推，但是CBO优化器打开可以进行谓词下推
       情况3：过滤条件写在on后面，并且过滤字段属于保留表，不管CBO优化器是否打开，都不会进行谓词下推
       情况4：过滤条件写在on后面，但是过滤字段属于空行表，可以进行谓词下推

       建议：如果进行的是左连接，则右表的过滤条件尽量写在on后面，左表的条件尽量写在where后面。
       全外连接：
       情况1：过滤条件写在where后面，关闭CBO优化器则不会进行下推，打开CBO优化器会进行下推
       情况2：过滤条件写在on后面,不可以进行谓词下推

9.导致数据倾斜的问题：
   1.数据分组时候重复数据太多，则可能会导致数据侵袭则可以开启负载均衡
       2.数据处理的时候null值太多，在处理的数据的时候空值就会被分配到一个reduce中，则也可能导致数据倾斜。
       一般在一些日志文件中null值会比较多，因为在记录日志的时候没有抓取到的数据则都保存为null。

       解决办法：1.可以给null值在处理的时候设置一个随机数，则就可以将空值的数据随机分配到不同的reduce中处理
2.如果null值对后续的数据分析没有用则可以提前过滤掉。
3.数据类型不相同也可能会导致数据倾斜。
如果有两张表进行关联操作，关联的列的数据类型一个是数字类型一个是字符串类型，则会导致所有的字符串类型全部会被分配到一个reduce中则导致数据倾斜。
解决办法：提前将数据的数据类型进行统一。

10.合理的控制map和reduce的数量
   1.map的数量的控制
       map数量=文件的大小/split片的大小（128M）

       1.map端输入合并小文件：用来减少map的数量
       如果在map端有大量的小文件输入，则会导致系统会分配大量的map任务来进行梳理数据，就会造成资源的浪费，所以建议将小文件进行合并
           参数的设置：
           set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; -- Map端输入、合并文件之后按照block的大小分割（默认）
           set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat; -- Map端输入，不合并
       2.map端输出合并
set hive.merge.mapfiles=true; -- 是否合并Map输出文件, 默认值为真

       3.通过设置split的大小来控制map的数量，split越大map数量就越小，split越小则map数量就越多。
       split默认的大小128M
           设置split大小：
           set mapred.max.split.size=256000000 --集群默认值
           set mapred.min.split.size=10000000 --集群默认值

           set dfs.blocksize=134217728 设置block块的大小，默认128M
       4.直接通过参数的方式来设置map的数量
       set mapred.map.tasks=n; 直接设置map的数量
           注意：设置的n必须要大于默认的map数量才会生效。

       2.reduce数量的控制：
默认情况下reduce的数量由每一个reduce梳理的数量来决定的。
set hive.exec.reducers.bytes.per.reducer=256000000   --默认256M
       设置每一个mapreduce中最多启动的reduce的数量
       set hive.exec.reducers.bytes.per.reducer=1009

       直接设置reduce的数量：
       set mapred.reduce.tasks=n;
       set mapreduce.job.reduces=m;
       最终的reduce的数量max(n,m)

       reduce端进行合并输出
       set hive.merge.mapredfiles=true; -- 是否合并Reduce 端输出文件,默认值为假
       set hive.merge.size.per.task=25610001000; -- 合并文件的大小,默认值为 256000000

以下情况只会启动一个reduce
       1.使用order by排序
           2.使用聚合函数的时候没有使用groupby
           3.如果进行的是笛卡尔积操作也会只启动一个reduce。

3.hive架构层面优化：

1.本地化执行

         Hive在集群上查询时，默认是在集群上多台机器上运行，需要多个机器进行协调运行，这种方式很好的解决了大数据量的查询问题。但是在Hive查询处理的数据量比较小的时候，其实没有必要启动分布式模式去执行，因为以分布式方式执行设计到跨网络传输、多节点协调等，并且消耗资源。对于小数据集，可以通过本地模式，在单台机器上处理所有任务，执行时间明显被缩短。
       set hive.exec.mode.local.auto=true; -- 打开hive自动判断是否启动本地模式的开关
       set hive.exec.mode.local.auto.input.files.max=4; -- map任务数最大值
       set hive.exec.mode.local.auto.inputbytes.max=134217728; -- map输入文件最大大小

   2.JVM重用
       Hive 语句最终会转换为一系列的 MapReduce 任务，每一个MapReduce 任务是由一系列的Map Task 和 Reduce Task 组成的，默认情况下，MapReduce 中一个 Map Task 或者 Reduce Task 就会启动一个 JVM 进程，一个 Task 执行完毕后，JVM进程就会退出。
       这样如果任务花费时间很短，又要多次启动 JVM 的情况下，JVM的启动、关闭时间会变成一个比较大的消耗，这时，可以通过重用 JVM 来解决。
       set mapred.job.reuse.jvm.num.tasks=5;
       JVM也是有缺点的，开启JVM重用会一直占用使用到的 task 的插槽，以便进行重用，直到任务完成后才会释放。
       如果某个不平衡的job中有几个 reduce task 执行的时间要比其他的 reduce task 消耗的时间要多得多的话，那么保留的插槽就会一直空闲却无法被其他的 job 使用，直到所有的 task 都结束了才会释放。
3.并行执行
       有的查询语句，hive会将其转化为一个或多个阶段，包括：MapReduce 阶段、抽样阶段、合并阶段、limit 阶段等。默认情况下，
       一次只执行一个阶段。但是，如果某些阶段不是互相依赖，是可以并行执行的。多阶段并行是比较耗系统资源的。
       set hive.exec.parallel=true; -- 可以开启并发执行。
       set hive.exec.parallel.thread.number=16; -- 同一个sql允许最大并行度，默认为8。