菜鸟笔记
提升您的技术认知

Hive-on-Spark调优

阅读 : 1255

集群规划 

Yarn配置

yarn.nodemanager.resource.memory-mb

该参数的含义是,一个NodeManager节点分配给Container使用的内存。该参数的配置,取决于NodeManager所在节点的总内存容量和该节点运行的其他服务的数量。

考虑上述因素,此处可将该参数设置为64G,如下:

<property>
    <name>yarn.nodemanager.resource.memory-mb</name>
    <value>65536</value>
</property>

yarn.nodemanager.resource.cpu-vcores

该参数的含义是,一个NodeManager节点分配给Container使用的CPU核数。该参数的配置,同样取决于NodeManager所在节点的总CPU核数和该节点运行的其他服务。

考虑上述因素,此处可将该参数设置为16。尽量1cpu对应的4G内存

<property>
    <name>yarn.nodemanager.resource.cpu-vcores</name>
    <value>16</value>
</property>

yarn.scheduler.maximum-allocation-mb

该参数的含义是,单个Container能够使用的最大内存。由于Spark的yarn模式下,Driver和Executor都运行在Container中,故该参数不能小于Driver和Executor的内存配置,推荐配置如下:

<property>
    <name>yarn.scheduler.maximum-allocation-mb</name>
    <value>16384</value>
</property>

yarn.scheduler.minimum-allocation-mb

该参数的含义是,单个Container能够使用的最小内存,推荐配置如下:

<property>
    <name>yarn.scheduler.minimum-allocation-mb</name>
    <value>512</value>
</property>

Yarn配置实操

修改$HADOOP_HOME/etc/hadoop/yarn-site.xml文件

<property>
    <name>yarn.nodemanager.resource.memory-mb</name>
    <value>65536</value>
</property>
<property>
    <name>yarn.nodemanager.resource.cpu-vcores</name>
    <value>16</value>
</property>
<property>
    <name>yarn.scheduler.maximum-allocation-mb</name>
    <value>16384</value>
</property>
<property>
    <name>yarn.scheduler.minimum-allocation-mb</name>
    <value>512</value>
</property>

Spark配置

Executor配置说明

Executor CPU核数配置

单个Executor的CPU核数,由spark.executor.cores参数决定,建议配置为4-6,具体配置为多少,视具体情况而定,原则是尽量充分利用资源。

此处单个节点共有16个核可供Executor使用,则spark.executor.core配置为4最合适。原因是,若配置为5,则单个节点只能启动3个Executor,会剩余1个核未使用;若配置为6,则只能启动2个Executor,会剩余4个核未使用。

Executor内存配置

Spark在Yarn模式下的Executor内存模型如下图所示:

Executor相关的参数有:spark.executor.memory和spark.executor.memoryOverhead。spark.executor.memory用于指定Executor进程的堆内存大小,这部分内存用于任务的计算和存储;spark.executor.memoryOverhead用于指定Executor进程的堆外内存,这部分内存用于jvm的额外开销,操作系统开销等。两者的和才算一个Executor进程所需的总内存大小。默认情况下spark.executor.memoryOverhead的值等于spark.executor.memory*0.1。

以上两个参数的推荐配置思路是,先按照单个NodeManager的核数和单个Executor的核数,计算出每个NodeManager最多能运行多少个Executor。在将NodeManager的总内存平均分配给每个Executor,最后再将单个Executor的内存按照大约10:1的比例分配到spark.executor.memory和spark.executor.memoryOverhead。

根据上述思路,可得到如下关系:

(spark.executor.memory+spark.executor.memoryOverhead)= yarn.nodemanager.resource.memory-mb * (spark.executor.cores/yarn.nodemanager.resource.cpu-vcores)

经计算,此处应做如下配置:

spark.executor.memory    14G
spark.executor.memoryOverhead    2G

Executor个数配置

此处的Executor个数是指分配给一个Spark应用的Executor个数,Executor个数对于Spark应用的执行速度有很大的影响,所以Executor个数的确定十分重要。

一个Spark应用的Executor个数的指定方式有两种,静态分配动态分配

静态分配

可通过spark.executor.instances指定一个Spark应用启动的Executor个数。这种方式需要自行估计每个Spark应用所需的资源,并为每个应用单独配置Executor个数。

动态分配

动态分配可根据一个Spark应用的工作负载,动态的调整其所占用的资源(Executor个数)。这意味着一个Spark应用程序可以在运行的过程中,需要时,申请更多的资源(启动更多的Executor),不用时,便将其释放。

在生产集群中,推荐使用动态分配。动态分配相关参数如下:

#启动动态分配
spark.dynamicAllocation.enabled    true
#启用Spark shuffle服务
spark.shuffle.service.enabled    true
#Executor个数初始值
spark.dynamicAllocation.initialExecutors    1
#Executor个数最小值
spark.dynamicAllocation.minExecutors    1
#Executor个数最大值
spark.dynamicAllocation.maxExecutors    12
#Executor空闲时长,若某Executor空闲时间超过此值,则会被关闭
spark.dynamicAllocation.executorIdleTimeout    60s
#积压任务等待时长,若有Task等待时间超过此值,则申请启动新的Executor
spark.dynamicAllocation.schedulerBacklogTimeout    1s
#使用旧版的shuffle文件Fetch协议
spark.shuffle.useOldFetchProtocol true

说明:Spark shuffle服务的作用是管理Executor中的各Task的输出文件,主要是shuffle过程map端的输出文件。由于启用资源动态分配后,Spark会在一个应用未结束前,将已经完成任务,处于空闲状态的Executor关闭。Executor关闭后,其输出的文件,也就无法供其他Executor使用了。需要启用Spark shuffle服务,来管理各Executor输出的文件,这样就能关闭空闲的Executor,而不影响后续的计算任务了。

Driver配置说明

Driver主要配置内存即可,相关的参数有spark.driver.memory和spark.driver.memoryOverhead。

spark.driver.memory用于指定Driver进程的堆内存大小,spark.driver.memoryOverhead用于指定Driver进程的堆外内存大小。默认情况下,两者的关系如下:spark.driver.memoryOverhead=spark.driver.memory*0.1。两者的和才算一个Driver进程所需的总内存大小。

一般情况下,按照如下经验进行调整即可:假定yarn.nodemanager.resource.memory-mb设置为X,

若X>50G,则Driver可设置为12G,

若12G<X<50G,则Driver可设置为4G。

若1G<X<12G,则Driver可设置为1G。 

此处yarn.nodemanager.resource.memory-mb为64G,则Driver的总内存可分配12G,所以上述两个参数可配置为

spark.driver.memory    10G
spark.yarn.driver.memoryOverhead    2G

配置实操

修改spark-defaults.conf文件

修改$HIVE_HOME/conf/spark-defaults.conf

spark.master                               yarn
spark.eventLog.enabled                   true
spark.eventLog.dir    hdfs://myNameService1/spark-history
spark.executor.cores    4
spark.executor.memory    14g
spark.executor.memoryOverhead    2g
spark.driver.memory    10g
spark.driver.memoryOverhead    2g
spark.dynamicAllocation.enabled  true
spark.shuffle.service.enabled  true
spark.dynamicAllocation.executorIdleTimeout  60s
spark.dynamicAllocation.initialExecutors    1
spark.dynamicAllocation.minExecutors  1
spark.dynamicAllocation.maxExecutors  11
spark.dynamicAllocation.schedulerBacklogTimeout 1s
spark.shuffle.useOldFetchProtocol    true

修改$HIVE_HOME/conf/hive-site.xml配置文件

<property>
    <name>spark.dynamicAllocation.enabled</name>
    <value>true</value>
</property>

配置Spark shuffle服务

Spark Shuffle服务的配置因Cluster Manager(standalone、Mesos、Yarn)的不同而不同。此处以Yarn作为Cluster Manager。

(1)拷贝$SPARK_HOME/yarn/spark-3.0.0-yarn-shuffle.jar到

$HADOOP_HOME/share/hadoop/yarn/lib

(2)分发$HADOOP_HOME/share/hadoop/yarn/lib/yarn/spark-3.0.0-yarn-shuffle.jar

(3)修改$HADOOP_HOME/etc/hadoop/yarn-site.xml文件

<property>
    <name>yarn.nodemanager.aux-services</name>
    <value>mapreduce_shuffle,spark_shuffle</value>
</property>

<property>
    <name>yarn.nodemanager.aux-services.spark_shuffle.class</name>
    <value>org.apache.spark.network.yarn.YarnShuffleService</value>
</property>

(4)分发$HADOOP_HOME/etc/hadoop/yarn-site.xml文件

(5)重启Yarn

Hive SQL执行计划

Hive SQL的执行计划,可由Explain查看。

Explain呈现的执行计划,由一系列Stage组成,这一系列Stage具有依赖关系,每个Stage对应一个MapReduce Job或者Spark Job,或者一个文件系统操作等。

每个Stage由一系列的Operator组成,一个Operator代表一个逻辑操作,例如TableScan Operator,Select Operator,Join Operator等。

Stage与Operator的对应关系如下图:

分组聚合优化 

select
    coupon_id,
    count(*)
from dwd_trade_order_detail_inc
where dt='2020-06-16'
group by coupon_id;

优化前执行计划

优化思路 

优化思路为map-side聚合。所谓map-side聚合,就是在map端维护一个hash table,利用其完成分区内的、部分的聚合,然后将部分聚合的结果,发送至reduce端,完成最终的聚合。map-side聚合能有效减少shuffle的数据量,提高分组聚合运算的效率。

map-side 聚合相关的参数如下:

--启用map-side聚合
set hive.map.aggr=true;
--hash map占用map端内存的最大比例
set hive.map.aggr.hash.percentmemory=0.5;

优化后执行计划

Join优化

Hive Join算法概述

Hive拥有多种join算法,包括common join,map join,bucket map join等。下面对每种join算法做简要说明:

common join

Map端负责读取参与join的表的数据,并按照关联字段进行分区,将其发送到Reduce端,Reduce端完成最终的关联操作。

map join 

若参与join的表中,有n-1张表足够小,Map端就会缓存小表全部数据,然后扫描另外一张大表,在Map端完成关联操作。

bucket map join

若参与join的表均为分桶表,且关联字段为分桶字段,且大表的分桶数量是小表分桶数量的整数倍。此时,就可以以分桶为单位,为每个Map分配任务了,此时Map端就无需再缓存小表的全表数据了,而只需缓存其所需的分桶。

Map Join优化

select
    *
from
(
    select
        *
    from dwd_trade_order_detail_inc
    where dt='2020-06-16'
)fact
left join
(
    select
        *
    from dim_sku_full
    where dt='2020-06-16'
)dim
on fact.sku_id=dim.id;

优化前执行计划

优化思路

上述参与join的两表一大一小,可考虑map join优化。

Map Join相关参数如下:

--启用map join自动转换
set hive.auto.convert.join=true;
--common join转map join小表阈值
set hive.auto.convert.join.noconditionaltask.size=1xxxx;

优化后执行计划

数据倾斜优化 

数据倾斜说明

数据倾斜问题,通常是指参与计算的数据分布不均,即某个key或者某些key的数据量远超其他key,导致在shuffle阶段,大量相同key的数据被发往一个Reduce,进而导致该Reduce所需的时间远超其他Reduce,成为整个任务的瓶颈。

Hive中的数据倾斜常出现在分组聚合和join操作的场景中,下面分别介绍在上述两种场景下的优化思路。

分组聚合导致的数据倾斜

select
    province_id,
    count(*)
from dwd_trade_order_detail_inc
where dt='2020-06-16'
group by province_id;

优化前执行计划

优化思路

由分组聚合导致的数据倾斜问题主要有以下两种优化思路:

启用map-side聚合 

--启用map-side聚合
set hive.map.aggr=true;
--hash map占用map端内存的最大比例
set hive.map.aggr.hash.percentmemory=0.5;

启用map-side聚合后的执行计划如下图所示

启用skew groupby优化 

--启用分组聚合数据倾斜优化
set hive.groupby.skewindata=true;

join导致的数据倾斜

优化思路

由join导致的数据倾斜问题主要有以下两种优化思路:

使用map join

--启用map join自动转换
set hive.auto.convert.join=true;
--common join转map join小表阈值
set hive.auto.convert.join.noconditionaltask.size

使用map join优化后执行计划如下图

启用skew join优化

其原理如下图

--启用skew join优化
set hive.optimize.skewjoin=true;
--触发skew join的阈值,若某个key的行数超过该参数值,则触发
set hive.skewjoin.key=100000;

需要注意的是,skew join只支持Inner Join。

启动skew join优化后的执行计划如下图所示:

任务并行度优化

Map阶段并行度

Map端的并行度,也就是Map的个数。是由输入文件的切片数决定的。一般情况下,Map端的并行度无需手动调整。Map端的并行度相关参数如下:

--可将多个小文件切片,合并为一个切片,进而由一个map任务处理
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; 
--一个切片的最大值
set mapreduce.input.fileinputformat.split.maxsize=256000000;

Reduce阶段并行度

Reduce端的并行度,相对来说,更需要关注。默认情况下,Hive会根据Reduce端输入数据的大小,估算一个Reduce并行度。但是在某些情况下,其估计值不一定是最合适的,此时则需要人为调整其并行度。

Reduce并行度相关参数如下:

--指定Reduce端并行度,默认值为-1,表示用户未指定
set mapreduce.job.reduces;
--Reduce端并行度最大值
set hive.exec.reducers.max;
--单个Reduce Task计算的数据量,用于估算Reduce并行度
set hive.exec.reducers.bytes.per.reducer;

Reduce端并行度的确定逻辑为,若指定参数mapreduce.job.reduces的值为一个非负整数,则Reduce并行度为指定值。否则,Hive会自行估算Reduce并行度,估算逻辑如下:

假设Reduce端输入的数据量大小为totalInputBytes

参数hive.exec.reducers.bytes.per.reducer的值为bytesPerReducer

参数hive.exec.reducers.max的值为maxReducers

则Reduce端的并行度为:

其中,Reduce端输入的数据量大小,是从Reduce上游的Operator的Statistics(统计信息)中获取的。为保证Hive能获得准确的统计信息,需配置如下参数:

--执行DML语句时,收集表级别的统计信息
set hive.stats.autogather=true;
--执行DML语句时,收集字段级别的统计信息
set hive.stats.column.autogather=true;
--计算Reduce并行度时,从上游Operator统计信息获得输入数据量
set hive.spark.use.op.stats=true;
--计算Reduce并行度时,使用列级别的统计信息估算输入数据量
set hive.stats.fetch.column.stats=true;

小文件合并优化

优化说明

小文件合并优化,分为两个方面,分别是Map端输入的小文件合并,和Reduce端输出的小文件合并。

Map端输入文件合并

合并Map端输入的小文件,是指将多个小文件划分到一个切片中,进而由一个Map Task去处理。目的是防止为单个小文件启动一个Map Task,浪费计算资源。

相关参数为:

--可将多个小文件切片,合并为一个切片,进而由一个map任务处理
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; 

Reduce输出文件合并

合并Reduce端输出的小文件,是指将多个小文件合并成大文件。目的是减少HDFS小文件数量。

相关参数为:

--开启合并Hive on Spark任务输出的小文件
set hive.merge.sparkfiles=true;

其他优化

 CBO优化

优化说明

CBO是指Cost based Optimizer,即基于计算成本的优化。

在Hive中,计算成本模型考虑到了:数据的行数、CPU、本地IO、HDFS IO、网络IO等方面。Hive会计算同一SQL语句的不同执行计划的计算成本,并选出成本最低的执行计划。目前CBO在hive中主要用于join的优化,例如多表join的join顺序。

相关参数为:

--是否启用cbo优化 
set hive.cbo.enable=true;

关闭CBO优化

--关闭cbo优化 
set hive.cbo.enable=false;
--为了测试效果更加直观,关闭map join自动转换
set hive.auto.convert.join=false;

谓词下推

谓词下推(predicate pushdown)是指,尽量将过滤操作前移,以减少后续计算步骤的数据量。

相关参数为:

--是否启动谓词下推(predicate pushdown)优化
set hive.optimize.ppd = true;

需要注意的是:

CBO优化也会完成一部分的谓词下推优化工作,因为在执行计划中,谓词越靠前,整个计划的计算成本就会越低

矢量化查询

Hive的矢量化查询优化,依赖于CPU的矢量化计算,CPU的矢量化计算的基本原理如下图:

Hive的矢量化查询,可以极大的提高一些典型查询场景(例如scans, filters, aggregates, and joins)下的CPU使用效率。

相关参数如下:

set hive.vectorized.execution.enabled=true;

 若执行计划中,出现“Execution mode: vectorized”字样,即表明使用了矢量化计算。

参考资料:

1.https://docs.cloudera.com/documentation/enterprise/6/6.3/topics/admin_hos_tuning.html#hos_tuning

2.Hive on Spark: Getting Started - Apache Hive - Apache Software Foundation