最实用的hive优化参数配置，session级别配置灵活性高_hive session

前言

在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
) ;
1
2
3
4
5
6

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
1
2
3
4

以上的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上允许创建的分区数量设置
1
2
3

6 设置reduce数量

set mapred.reduce.tasks= 20;
1

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; 
1
2
3
4

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;
  }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

protected long computeSplitSize(long blockSize, long minSize,
                                  long maxSize) {
    return Math.max(minSize, Math.min(maxSize, blockSize));
  }
1
2
3
4

8 hive.new.job.grouping.set.cardinality 默认值是30，也就是在GROUPING SETS里面集合数据超过这个值的时候，会增加一个job来降低mapreduce的处理。使用with cube 、grouping set的优化配置

set hive.new.job.grouping.set.cardinality = 4100;
1

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;
1
2
3
4
5
6
7
8

10 在group by 阶段出现数据倾斜应该设置为true， 若group的键对应的记录条数超过100000就会进行此优化

set hive.groupby.skewindata=true;
 set hive.groupby.mapaggr.checkinterval=100000
1
2

默认该参数的值为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
1
2
3
4

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
1
2

reduce端的大小设置：

set mapreduce.reduce.memory.mb=10000;
set mapreduce.reduce.java.opts=-Xmx10000m;
1
2

12 控制是否启用查询执行的向量模式

set hive.vectorized.execution.enabled=false;
1