HiveSQL运行优化参数配置

HiveSQL常用数据处理语句

HiveSQL中复杂数据类型操作

我们平时在使用hive执行一些SQL任务时。经常会遇到执行速度很慢，数据倾斜，资源不够用等问题。那是因为我们没有合理的使用hive。hive 的主要配置文件为conf中hive-site.xml，里面包含许多配置参数，灵活的根据业务进行相关的参数配置，可以解决以上问题。下面将介绍hive的全部参数的意义以及如何配置。

常规优化配置

Map的任务数配置

Reduce的任务数配置

其他可选配置

常规优化配置

--hive任务配置参数
set mapreduce.job.queuename=default;
set mapred.job.name=one_big_job;set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
SET hive.exec.max.dynamic.partitions=1000000;
SET hive.exec.max.dynamic.partitions.pernode=1000000;
set hive.exec.max.created.files=6553500;set mapred.max.split.size=202400000;
set mapred.min.split.size.per.node=202400000;
set mapred.min.split.size.per.rack=202400000;set mapred.reduce.tasks=200;
set hive.exec.reducers.max=200;set hive.exec.parallel=true;
set hive.exec.parallel.thread.number=24;set hive.groupby.skewindata=true;
set hive.map.aggr=true;

以上参数可以直接配置使用，详细的参数解释及最优数值大小参考如下。

Map的任务数配置

执行任务时Map的任务数配置，即执行任务时，上图标红的，number of mappers: 1 的数量。

map的个数设置其实对执行效率有很大的影响：

如果mappers数量过多，map任务启动和初始化的时间远远大于逻辑处理的时间，就会造成很大的资源浪费。而且，同时可执行的map数是受限的；
如果mappers数量过少，Hadoop的计算资源没有充分的利用，计算缓慢；

map的个数主要的决定因素有： input的文件总个数，input的文件大小，集群设置的文件块大小(默认为128M)

map个数的计算公式如下：splitsize=max(minimumsize,min(maximumsize,blocksize))

如果没有设置minimumsize和maximumsize，splitsize的大小默认等于blocksize

计算过程可以简化为如下的公式，详细算法可以参照FileInputSplit类中的getSplits方法

total_split ;
for(file ：输入目录中的每个文件)
{
file_split = 1;
if(file.size>splitsize)
{
file_split=file_size/splitsize;
}
total_split+=file_split;
}

由于mapreduce中没有办法直接控制map数量，所以只能通过设置每个map中处理的数据量进行设置；reduce是可以直接设置的

set mapred.max.split.size = 256000000; （200M = 200 * 1000 * 1000）
-----决定每个map处理的最大的文件大小，单位为B

set mapred.min.split.size.per.node = 256000000;
-----节点中可以处理的最小的文件大小

set mapred.min.split.size.per.rack = 256000000;
-----机架中可以处理的最小的文件大小

Reduce的任务数配置

执行任务时 reducers 的任务数配置，即执行任务时，上图标红的，number of reducers: 159 的数量。

影响：reduce的个数设置其实对执行效率有很大的影响：

如果reduce太少，如果数据量很大，会导致这个reduce异常的慢，从而导致这个任务不能结束，也有可能会OOM；
如果reduce太多，产生的小文件太多，合并起来代价太高，namenode的内存占用也会增大；

reduce个数可以由以下三个参数直接决定:

set mapred.reduce.tasks=200;
-----这个参数如果指定了，hive就不会用它的estimation函数来自动计算reduce的个数，而是用这个参数来启动reducer。默认是-1，根据计算数量的大小来设定，一般设置为200，执行任务是reduce任务数即为200；

set hive.exec.reducers.bytes.per.reducer = 500000000；（500M = 500 * 1000 * 1000）
-----默认是1G，输入文件如果是10G，那么就会起10个reducer,，单位为B;

set hive.exec.reducers.max=200;
-----这个参数控制最大的reducer的数量，如果输入文件大小/hive.exec.reducers.bytes.per.reducer 所得出的大小 max 则会启动这个参数所指定的reduce个数；这个并不会影响mapre.reduce.tasks参数的设置，默认的max是999；

其他可选配置

常规配置

set mapreduce.job.queuename=default;
-----设置任务运行队列，同一个Hadoop集群不同的队列资源的相互隔离的，避免因大任务计算导致资源耗竭；

set mapred.job.name=one_big_job;
------设置任务运行任务名称，在yarn的任务日志页面可以现在此 Jobname；

set mapreduce.task.timeout = 60000
-----mr程序的task执行情况汇报过期时间，默认60000(10分钟)，设置为0表示不进行该值的判断。

set hive.cli.print.current.db=true;
-----让提示符显示当前库

set hive.cli.print.header=true;
-----显示查询结果时显示字段名称：

动态分区插入数据参数配置

set hive.exec.dynamic.partition=true;
-----是开启动态分区，即动态插入分区，可以根据数据计算结果值为分区；

set hive.exec.dynamic.partition.mode=nonstrict;
-----这个属性默认值是strict，就是要求分区字段必须有一个是静态的分区值；

set hive.exec.max.dynamic.partitions=1000000;
-----每一个mapreduce job允许创建的分区的最大数量，如果超过了这个数量就会报错，缺省值100；

set hive.exec.max.dynamic.partitions.pernode=1000000;
-----一个dml语句允许创建的所有分区的最大数量，缺省值1000；

set hive.exec.max.created.files=6553500;
-----所有的mapreduce job允许创建的文件的最大数量，缺省值100000；

作业有多个可并行的job时，设置任务并行及并行个数：

set hive.exec.parallel=true;
-----开启任务并行执行

set hive.exec.parallel.thread.number=8;
-----同一个sql允许并行任务的最大线程数

针对一条HiveSQL中存在的查询操作无直接关联，可以并行执行，如union all操作，可以开启并行执行；left join类的操作则语句配置失效，本身不能进行并行执行，存在依赖；

多条独立HiveSQL需要并行执行，可以同时提交任务，与HvieSQL这里的配置没有关系；

配置任务执行引擎

set hive.execution.engine=mr; （默认）

set hive.execution.engine=tez;（需要Hadoop集群安装/支持）

set hive.execution.engine=spark;

如果设置执行引擎为mr，那么就会调用Hadoop的maprecude来运行需要执行job的程序；

如果设置执行引擎为spark，那么就会调用spark来执行任务。有条件的话，就设置执行引擎为Spark，运行的比Hadoop的MapReduce快了很多。

如果设置执行引擎为tez，Tez是一个Hive的运行引擎（需要先安装Tez），性能优于MR，Tez基于内存的计算使得hive可以有更高的运算效率。

小表数据关联计算使用mapjoin

set hive.auto.convert.join = true;
-----是否自动转换为mapjoin

set hive.mapjoin.smalltable.filesize=50000000;
-----小表的最大文件大小，默认为25*000*000，即25M

set hive.auto.convert.join.noconditionaltask = true;
-----是否将多个mapjoin合并为一个

set hive.auto.convert.join.noconditionaltask.size = 10000000; （10*1000*1000）
-----多个mapjoin转换为1个时，所有小表的文件大小总和的最大值。

MapJoin顾名思义，就是在Map阶段进行表之间的连接。而不需要进入到Reduce阶段才进行连接。这样就节省了在Shuffle阶段时要进行的大量数据传输。从而起到了优化作业的作用。

MapJoin适用的场景：mapjoin的适用场景如关联操作中有一张表非常小，.不等值的链接操作。通过上面分析你会发现，并不是所有的场景都适合用MapJoin. 它通常会用在如下的一些情景：在二个要连接的表中，有一个很大，有一个很小，这个小表可以存放在内存中而不影响性能。这样我们就把小表文件复制到每一个Map任务的本地，再让Map把文件读到内存中待用。

数据倾斜

set hive.groupby.skewindata=true;
-----有数据倾斜的时候进行负载均衡，当选项设定为 true，生成的查询计划会有两个 MR Job。第一个 MR Job 中，Map 的输出结果集合会随机分布到 Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；第二个 MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中），最后完成最终的聚合操作。

set hive.map.aggr=true;
-----在map中会做部分聚集操作，效率更高但需要更多的内存。

如果以上不管用，可以对倾斜的数据进行单独的sql处理。

内存溢出优化

Hadoop处理数据时，出现内存溢出的处理方法（内存调优），Mapper/Reducer阶段JVM内存溢出，堆内存不足时，一般会抛出如下几种异常：

java.lang.OutOfMemoryError:” GC overhead limit exceeded

Error: Java heapspace

running beyondphysical memory limits.Current usage: 4.3 GB of 4.3 GBphysical memoryused; 7.4 GB of 13.2 GB virtual memory used. Killing container

Exception: java.lang.OutOfMemoryError thrown from theUncaughtExceptionHandler in thread

Socket Reader #1 for port 30703

Halting due to Out Of Memory Error...

Halting due to Out Of Memory Error...

Halting due to Out Of Memory Error...

java.lang.OutOfMemoryError:Direct buffffer memory

目前MapReduce主要通过两个组参数去控制内存：（将如下参数调大）

mapreduce.map.java.opts=-Xmx2048m
-----默认参数，表示jvm堆内存,注意是mapreduce不是mapred

mapreduce.map.memory.mb=2304
-----container的内存

mapreduce.reduce.java.opts=-Xmx2048m
-----默认参数，表示jvm堆内存

mapreduce.reduce.memory.mb=2304
-----container的内存

HiveSQL执行命令

hive -e 'select * from area where code=202020' >> /root/files/a.csv

hive -f sqlfile.sql >> /files/result.csv

hive -hiveconf day=2015 -f sqlfile.sql >> /files/result.csv
-----传入一个参数

hive -hiveconf year=2015 -hiveconf month=09 -hiveconf day=11 -f sqlfile.sql >> /files/result.csv
-----传入多个参数

nohub hive -f sqlfile.sh >> result.csv 2>&1 &
-----后台执行

Hive中引入外部包

hive默认的函数并不是太完整，以后我们使用的使用肯定需要自己补充一些自定义函数的。

ADD jar hdfs://ip:9000/user/hadoop/share/HiveUdf.jar;
-----在hive中添加包

delete jar hdfs://ip:9000/user/hadoop/share/HiveUdf.jar;
-----在hive中删除包

CREATE TEMPORARY FUNCTION tmp_fun as 'com.hive.udf.Encry';
-----在hive中创建函数

select tmp_fun('123');
-----在hive中使用函数

Hive中数据压缩

中间压缩处理hive查询的多个job之间的数据，对于中间压缩，最好选择一个节省cpu耗时的压缩方式
set hive.exec.compress.intermediate=true;
set hive.intermediate.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;
set hive.intermediate.compression.type=BLOCK;
-----按块压缩，而不是记录

最终输出压缩（选择压缩效果好的，减少储存空间）
set hive.exec.compress.output=true;
set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;
set mapred.output.compression.type=BLOCK;
-----按块压缩，而不是记录