前言
在Hive优化方面,要做到性能最优,那就是得定制优化,针对不同的sql脚本设置不同的参数,配置不同的map和reduce数。保证局部性能最优,结果才会是效率最高。
那么在定制优化方面使用session级别的配置就是对症下药。所有的设置的参数只在一次Hive的cli命令起作用,关闭窗口或者退出命令行后整个配置参数全部失效,不会对其他脚本的运行环境造成影响。
配置清单:
- 1 创建session级别临时表,会话关闭临时表自动销毁
- 2 设置程序运行的队列环境
- 3 设置sql语句查询结果是否显示列名
- 4 开启动态分区的设置
- 5 动态设置分区达到默认数量上限
- 6 设置reduce数量
- 7 设置map数量
- 8 使用with cube 、grouping set的优化配置
- 9 是否自动转为mapjoin
- 10 在group by 阶段出现数据倾斜
- 11 hadoop作业JVM堆大小设置
- 12 控制是否启用查询执行的向量模式
详细讲解:
1 创建session级别临时表,会话关闭临时表自动销毁
CREATE TEMPORARY TABLE tmp
(
a string
,b string
,c string
) ;
2 设置程序运行的队列环境
set mapred.job.queue.name=ne;
3 设置sql语句查询结果是否显示列名
set hive.cli.print.header=true;
select * from tmp2_jc_test limit 10;
4 开启动态分区的设置
set hive.exec.dynamic.partition.mode=nonstrict;
INSERT overwrite table tmp partition(etl_date)
select
a,b,c,table_column4_date from tmp2
以上的tmp表会按照 table_column4_date 字段进行动态创建分区
5 动态设置分区是有默认数量上限的,如果数量超出了默认设置就需要配置下面的参数
SET hive.exec.dynamic.partition=true;
SET hive.exec.max.dynamic.partitions=2048; --自动创建分区的数量设置
SET hive.exec.max.dynamic.partitions.pernode=2048; --一个node上允许创建的分区数量设置
6 设置reduce数量
set mapred.reduce.tasks= 20;
7 设置map数量
set mapred.max.split.size=5000000;
set mapred.min.split.size.per.node=5000000;
set mapred.min.split.size.per.rack=5000000;
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
map数量是不能直接设定一个固定值。而是通过hadoop内置的算法计算出来的数值。
主要的决定因素有: input的文件总个数,input的文件大小,集群设置的文件块大小(目前为128M, 可在hive中通过set dfs.block.size;命令查看到,该参数不能自定义修改);
map任务要划分的大小(splitSize )=(maxSize与blockSize之间的最小值)与minSize之间的最大值
bytesRemaining 是单个输入源文件未划分的字节数
根据getSplits方法,map任务划分的数量=输入源文件数目 * (bytesRemaining / splitSize个划分任务+bytesRemaining不能被splitSize 整除的剩余大小单独划分一个任务 )
个人理解: 可以根据输入的文件大小、文件个数以及集群中block大小就大致可以推算出大概的map数,但是,map和reduce数不是越多越好,也不是越少越好。map和reduce数根据集群的节点数而定。尽量让计算任务平均分摊到每个节点上,才能达到最高的计算效率
hive 源码如下:FileInputFormat.java
public List<InputSplit> getSplits(JobContext job) throws IOException {
StopWatch sw = new StopWatch().start();
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
long maxSize = getMaxSplitSize(job);
List<InputSplit> splits = new ArrayList<InputSplit>();
List<FileStatus> files = listStatus(job);
for (FileStatus file: files) {
Path path = file.getPath();
long length = file.getLen();
if (length != 0) {
BlockLocation[] blkLocations;
if (file instanceof LocatedFileStatus) {
blkLocations = ((LocatedFileStatus) file).getBlockLocations();
} else {
FileSystem fs = path.getFileSystem(job.getConfiguration());
blkLocations = fs.getFileBlockLocations(file, 0, length);
}
if (isSplitable(job, path)) {
long blockSize = file.getBlockSize();
long splitSize = computeSplitSize(blockSize, minSize, maxSize);
long bytesRemaining = length;
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
splits.add(makeSplit(path, length-bytesRemaining, splitSize,
blkLocations[blkIndex].getHosts(),
blkLocations[blkIndex].getCachedHosts()));
bytesRemaining -= splitSize;
}
if (bytesRemaining != 0) {
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
blkLocations[blkIndex].getHosts(),
blkLocations[blkIndex].getCachedHosts()));
}
} else { // not splitable
splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
blkLocations[0].getCachedHosts()));
}
} else {
//Create empty hosts array for zero length files
splits.add(makeSplit(path, 0, length, new String[0]));
}
}
// Save the number of input files for metrics/loadgen
job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
sw.stop();
if (LOG.isDebugEnabled()) {
LOG.debug("Total # of splits generated by getSplits: " + splits.size()
+ ", TimeTaken: " + sw.now(TimeUnit.MILLISECONDS));
}
return splits;
}
protected long computeSplitSize(long blockSize, long minSize,
long maxSize) {
return Math.max(minSize, Math.min(maxSize, blockSize));
}
8 hive.new.job.grouping.set.cardinality 默认值是30,也就是在GROUPING SETS里面集合数据超过这个值的时候,会增加一个job来降低mapreduce的处理。使用with cube 、grouping set的优化配置
set hive.new.job.grouping.set.cardinality = 4100;
9 是否自动转为mapjoin
--是否自动转换为mapjoin
set hive.auto.convert.join = true;
--小表的最大文件大小,默认为25000000,即25M
set hive.mapjoin.smalltable.filesize = 25000000;
--是否将多个mapjoin合并为一个
set hive.auto.convert.join.noconditionaltask = true;
--多个mapjoin转换为1个时,所有小表的文件大小总和的最大值。
set hive.auto.convert.join.noconditionaltask.size = 10000000;
10 在group by 阶段出现数据倾斜应该设置为true, 若group的键对应的记录条数超过100000就会进行此优化
set hive.groupby.skewindata=true;
set hive.groupby.mapaggr.checkinterval=100000
默认该参数的值为false,表示不启用,要启用时,可以set hive.groupby.skewindata=ture;进行启用。
当启用时,能够解决数据倾斜的问题,但如果要在查询语句中对多个字段进行去重统计时会报错
hive> set hive.groupby.skewindata=true;
hive> select count(distinct id),count(distinct x) from test;
FAILED: SemanticException [Error 10022]: DISTINCT on different columns not supported with skew in data
11 hadoop作业JVM堆大小设置
map端的大小设置:
set mapreduce.map.java.opts=-Xmx4096m;--默认值:-Xmx200。 map/reduce执行参数设置,可以参数Hadoop Mapreduce Error: GC overhead limit exceeded
set mapreduce.map.memory.mb=4096; -- map内存申请大小,单位m,默认值 1024
reduce端的大小设置:
set mapreduce.reduce.memory.mb=10000;
set mapreduce.reduce.java.opts=-Xmx10000m;
12 控制是否启用查询执行的向量模式
set hive.vectorized.execution.enabled=false;