Hive On Spark性能调优

一、基础配置

我们公司yarn node节点的可用资源配置为：单台node节点可用资源数：核数33cores、内存110G。Hive on Spark任务的基础配置，主要配置对象包括：Executor和Driver内存，Executor配额，任务并行度。

1、Executor内存和核数

配置参数为spark.executor.memory和spark.executor.cores。如果要最大化使用core，建议将core设置为4、5、6，且满足core的个数尽量可以整除yarn资源核数。yarn资源可用33核，建议spark.executor.cores设置为4，最多剩下一个core，如果设置为5，6都会有3个core剩余。 spark.executor.cores=4，由于总共有33个核，那么最大可以申请的executor数是8。总内存处以8，也即是 110/8，可以得到每个executor约13.75GB内存。

建议 spark.executor.memoryOverhead（spark的executor堆外内存）站总内存的 15%-20%。那么最终 spark.executor.memoryOverhead=2.75 G 和spark.executor.memory=11 G
注意：默认情况下 spark.executor.memoryOverhead = max(executorMemory * 0.10, 384M)，正常情况下不需要手动设置spark堆外内存，如果spark任务出现如下报错，可以手动提高堆外内存大小。

注意：默认情况下 spark.executor.memoryOverhead = max(executorMemory * 0.10, 384M)，正常情况下不需要手动设置spark堆外内存，如果spark任务出现如下报错，可以手动提高堆外内存大小。

Container killed by YARN for exceeding memory limits. 16.9 GB of 16 GB physical memory used. Consider boosting spark.yarn.executor.memoryOverhead.

2、Driver内存

对于drvier的内存配置，主要有两个参数：

spark.driver.memoryOverhead 每个driver能从yarn申请的堆外内存的大小。
spark.driver.memory 当运行hive on spark的时候，每个spark driver能申请的最大jvm 堆内存。该参数结合 spark.driver.memoryOverhead共同决定着driver的内存大小。

Driver的内存通常来说不设置，或者设置1G左右应该就够了。需要注意的是，如果需要使用collect算子将RDD的数据全部拉取到Driver端进行处理，那么必须确保Driver的内存足够大，否则会出现OOM内存溢出的问题。

3、Executor个数

配置参数为spark.executor.instances。该参数用于设置Spark作业总共要用多少个Executor进程来执行。

executor的数目是由每个节点运行的executor数目和集群的节点数共同决定。我们离线集群27个节点，那么离线spark任务使用的最大executor数就是 216(27*8). 最大数目可能比这个小点，因为driver也会消耗核数和内存。

该参数可以结合spark.executor.cores设置，默认单个spark任务最大不超过60cores，spark.executor.cores设置为4，则spark.executor.instances不超过15。

4、并行度

设置spark任务的并行度参数为spark.default.parallelism。spark任务每个stage的task个数=max(spark.default.parallelism, HDFS的block数量)。如果不设置该参数，Spark自己根据底层HDFS的block数量来设置task的数量，默认是一个HDFS block对应一个task。spark默认spark.default.parallelism配置较少，如果task个数比较少的话，前面spark资源配置没有意义。官网建议：该参数设置为 num-executors * executor-cores的2~3倍较为合适。

5、参数配置

SET hive.execution.engine=spark;
-- 修改spark on yarn的资源队列
SET spark.yarn.queue=root.dw_offline_day;
-- 设置Driver内存
SET spark.driver.memory=1G;
-- 设置Executor核数
SET spark.executor.cores=4;
-- 设置Executor内存
SET spark.executor.memory=11G;
-- 设置Executor个数
SET spark.executor.instances=10;
-- 设置spark任务并行度
SET spark.default.parallelism=120;

二、动态资源分配

当一个运行时间比较长的spark任务，如果分配给他多个Executor，可是却没有task分配给它，而此时有其他的yarn任务资源紧张，这就造成了很大的资源浪费和资源不合理的调度。动态资源调度就是为了解决这种场景，根据当前应用任务的负载情况，实时的增减Executor个数，从而实现动态分配资源，使整个Spark系统更加健康。

1、资源分配策略

开启spark动态资源分配后，application会在task因没有足够资源被挂起的时候去动态申请资源。当任务挂起或等待spark.dynamicAllocation.schedulerBacklogTimeout(默认1s)的时间后，会开始动态资源分配；之后每隔spark.dynamicAllocation.sustainedSchedulerBacklogTimeout(默认1s)时间申请一次，直到申请到足够的资源。每次申请的资源量是指数增长的，即1,2,4,8等。

2、资源回收策略

当application的executor空闲时间超过spark.dynamicAllocation.executorIdleTimeout（默认60s）后，就会被回收。

3、参数配置

SET hive.execution.engine=spark;
-- 修改spark on yarn的资源队列
SET spark.yarn.queue=root.dw_offline_day;
-- 设置Driver内存
SET spark.driver.memory=1G;
-- 设置Executor核数
SET spark.executor.cores=4;
-- 设置Executor内存
SET spark.executor.memory=8G;
-- 开启动态资源分配
SET spark.dynamicAllocation.enabled=true;
-- 每个Application最小分配的executor数
SET spark.dynamicAllocation.minExecutors=1;
-- 如果所有的executor都移除了，重新请求时启动的初始executor数
SET spark.dynamicAllocation.initialExecutors=1;
-- 每个Application最大并发分配的executor数
SET spark.dynamicAllocation.maxExecutors=15;
-- 有新任务处于等待状态，并且等待时间超过该时间段，会依次启动executor, 每次启动1,2,4,8...个executor
SET spark.dynamicAllocation.schedulerBacklogTimeout=1s;
-- 动态启动executor的间隔
SET spark.dynamicAllocation.sustainedSchedulerBacklogTimeout=5s;

三、小文件合并

1、参数配置

SET hive.merge.mapfiles=true;
SET hive.merge.mapredfiles=false;
SET hive.merge.sparkfiles=true;
-- 当输出文件的平均大小小于该值时，启动单独的任务进行文件merge
SET hive.merge.smallfiles.avgsize=16000000;
-- 合并文件的大小（默认256M）
SET hive.merge.size.per.task=256000000;

四、动态分区使用

使用场景：同一个SQL语句需要同时更新多个分区，类似于如下SQL语句：

INSERT OVERWRITE TABLE table_name PARTITION (dt)
SELECT id, name, dt
FROM ...

1、参数说明

（1）hive.exec.dynamic.partition

默认值：false
说明：是否启动动态分区

（2）hive.exec.dynamic.partition.mode

默认值：strict
说明：打开动态分区后，动态分区的模式，有strict和nonstrict两种模式，strict要求至少包含一个静态分区列，nonstrict则无此要求。

（3）hive.exec.max.dynamic.partitions

默认值：100000
说明：允许的最大的动态分区的个数。可以手动增加分区。

（4）hive.exec.max.dynamic.partitions.pernode

默认值：100
说明：在每个执行任务的节点上，最大可以创建多少个动态分区。

（5）hive.exec.default.partition.name

默认值： HIVE_DEFAULT_PARTITION
说明：在hive里面表可以创建成分区表，但是当分区字段的值是’’ 或者 null时，hive会自动将分区命名为默认分区名称。默认情况下，默认分区的名称为HIVE_DEFAULT_PARTITION，默认分区名称是可配置的。

2、参数配置

SET hive.exec.dynamic.partition=TRUE;
SET hive.exec.dynamic.partition.mode=nonstrict;
SET hive.exec.max.dynamic.partitions=100000;
SET hive.exec.max.dynamic.partitions.pernode=100000;

五、Hive参数配置

1、参数配置

-- 如果一个简单查询只包括一个group by和order by，此处可以设置为1或2。适用于模型层维度表处理
SET hive.optimize.reducededuplication.min.reducer=4;

-- 如果数据已经根据相同的key做好聚合，那么去除掉多余的map/reduce作业
SET hive.optimize.reducededuplication=true;

-- Map Join优化, 不太大的表直接通过map过程做join
SET hive.auto.convert.join=true;
SET hive.auto.convert.join.noconditionaltask=true;

-- Map Join任务HashMap中key对应value数量
SET hive.smbjoin.cache.rows=10000;

-- 可以被转化为HashMap放入内存的表的大小(官方推荐853M)
SET hive.auto.convert.join.noconditionaltask.size=200M;

-- 如果数据按照join的key分桶，hive将简单优化inner join(官方推荐关闭)
SET hive.optimize.bucketmapjoin= false;
SET hive.optimize.bucketmapjoin.sortedmerge=false;

-- 所有map任务可以用作Hashtable的内存百分比, 如果OOM, 调小这个参数(官方默认0.5)
SET hive.map.aggr.hash.percentmemory=0.5;

-- map端聚合(跟group by有关), 如果开启, Hive将会在map端做第一级的聚合, 会用更多的内存，开启这个参数 sum(1)会有类型转换问题
SET hive.map.aggr=false;

-- 将只有SELECT, FILTER, LIMIT转化为FETCH, 减少等待时间
SET hive.fetch.task.conversion=more;
SET hive.fetch.task.conversion.threshold=1073741824;

-- 新创建的表/分区是否自动计算统计数据
SET hive.stats.autogather=true;
SET hive.compute.query.using.stats=true;
SET hive.stats.fetch.column.stats=true;
SET hive.stats.fetch.partition.stats=true;

-- 在order by limit查询中分配给存储Top K的内存为10%
SET hive.limit.pushdown.memory.usage=0.1;

-- 是否开启自动使用索引
SET hive.optimize.index.filter=true;