文章目录

本课程所用集群由5台节点构成其中2台为master节点用于部署HDFS的NameNodeYarn的ResourceManager等角色另外3台为worker节点用于部署HDFS的DataNode、Yarn的NodeManager等角色。

1.2

hadoop100hadoop101hadoop102hadoop103hadoop104mastermasterworkerworkerworkerNameNodeNameNodeDataNodeDataNodeDataNodeResourceManagerResourceManagerNodeManagerNodeManagerNodeManagerJournalNodeJournalNodeJournalNodeZookeeperZookeeperZookeeperKafkaKafkaKafkaHiveserver2Metastorehive-clienthive-clienthive-clientSparkSparkSparkSparkDS-masterDS-masterDS-workerDS-workerDS-worderMaxwellmysqlflumeflume

第2章

需要调整的Yarn参数均与CPU、内存等资源有关核心配置参数如下。

1yarn.nodemanager.resource.memory-mb

该参数的含义是一个NodeManager节点分配给Container使用的内存。

该参数的配置取决于NodeManager所在节点的总内存容量和该节点运行的其他服务的数量。

yarn.nodemanager.resource.memory-mb

65536

2yarn.nodemanager.resource.cpu-vcores

该参数的含义是一个NodeManager节点分配给Container使用的CPU核数。

该参数的配置同样取决于NodeManager所在节点的总CPU核数和该节点运行的其他服务。

yarn.nodemanager.resource.cpu-vcores

3yarn.scheduler.maximum-allocation-mb

该参数的含义是单个Container能够使用的最大内存。

由于Spark的yarn模式下Driver和Executor都运行在Container中故该参数不能小于Driver和Executor的内存配置推荐配置如下

yarn.scheduler.maximum-allocation-mb

16384

4yarn.scheduler.minimum-allocation-mb

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

yarn.scheduler.minimum-allocation-mb

512

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

2修改如下参数

yarn.nodemanager.resource.memory-mb

65536

yarn.nodemanager.resource.cpu-vcores

yarn.scheduler.maximum-allocation-mb

16384

yarn.scheduler.minimum-allocation-mb

512

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

此处单个节点共有16个核可供Executor使用则spark.executor.core配置为4最合适。

原因是若配置为5则单个节点只能启动3个Executor会剩余1个核未使用若配置为6则只能启动2个Executor会剩余4个核未使用。

3.1.2

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.memoryspark.executor.memoryOverhead)

yarn.nodemanager.resource.memory-mb

(spark.executor.cores/yarn.nodemanager.resource.cpu-vcores)

经计算此处应做如下配置

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

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

1**静态分配**

可通过spark.executor.instances指定一个Spark应用启动的Executor个数。

这种方式需要自行估计每个Spark应用所需的资源并为每个应用单独配置Executor个数。

2**动态分配**

动态分配可根据一个Spark应用的工作负载动态的调整其所占用的资源Executor个数。

这意味着一个Spark应用程序可以在运行的过程中需要时申请更多的资源启动更多的Executor不用时便将其释放。

#启动动态分配

spark.dynamicAllocation.enabled

true

spark.dynamicAllocation.initialExecutors

#Executor个数最小值

spark.dynamicAllocation.minExecutors

#Executor个数最大值

spark.dynamicAllocation.maxExecutors

#Executor空闲时长若某Executor空闲时间超过此值则会被关闭

spark.dynamicAllocation.executorIdleTimeout

60s

#积压任务等待时长若有Task等待时间超过此值则申请启动新的Executor

spark.dynamicAllocation.schedulerBacklogTimeout

#spark

spark.shuffle.useOldFetchProtocol

true

shuffle服务的作用是管理Executor中的各Task的输出文件主要是shuffle过程map端的输出文件。

由于启用资源动态分配后Spark会在一个应用未结束前将已经完成任务处于空闲状态的Executor关闭。

Executor关闭后其输出的文件也就无法供其他Executor使用了。

需要启用Spark

shuffle服务来管理各Executor输出的文件这样就能关闭空闲的Executor而不影响后续的计算任务了。

3.2

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

spark.driver.memory用于指定Driver进程的堆内存大小spark.driver.memoryOverhead用于指定Driver进程的堆外内存大小。

默认情况下两者的关系如下spark.driver.memoryOverheadspark.driver.memory*0.1。

两者的和才算一个Driver进程所需的总内存大小。

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

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

spark.driver.memory

spark.yarn.driver.memoryOverhead

3.3

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

spark.master

hdfs://myNameService1/spark-history

spark.executor.cores

spark.dynamicAllocation.enabled

true

spark.dynamicAllocation.executorIdleTimeout

60s

spark.dynamicAllocation.initialExecutors

spark.dynamicAllocation.minExecutors

spark.dynamicAllocation.maxExecutors

spark.dynamicAllocation.schedulerBacklogTimeout

spark.shuffle.useOldFetchProtocol

true

Managerstandalone、Mesos、Yarn的不同而不同。

此处以Yarn作为Cluster

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文件

mapreduce_shuffle,spark_shuffle

yarn.nodemanager.aux-services.spark_shuffle.class

org.apache.spark.network.yarn.YarnShuffleService

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

5重启Yarn

Explain呈现的执行计划由一系列Stage组成这个Stage具有依赖关系每个Stage对应一个MapReduce

Job或者Spark

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

OperatorSelect

1.https://cwiki.apache.org/confluence/display/Hive/LanguageManualExplain

2.https://cwiki.apache.org/confluence/download/attachments/44302539/hos_explain.pdf?version1modificationDate1425575903211apiv2

第5章分组聚合优化

优化思路为map-side聚合。

所谓map-side聚合就是在map端维护一个hash

table利用其完成分区内的、部分的聚合然后将部分聚合的结果发送至reduce端完成最终的聚合。

map-side聚合能有效减少shuffle的数据量提高分组聚合运算的效率。

map-side

hive.map.aggr.hash.percentmemory0.5;

set

hive.groupby.mapaggr.checkinterval100000;

–map-side聚合所用的HashTable占用map任务堆内存的最大比例若超出该值则会对HashTable进行一次flush。

set

hive.map.aggr.hash.force.flush.memory.threshold0.9;

5.3

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

2****map

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

3****Sort

若参与join的表均为分桶表且关联字段为分桶字段且分桶字段是有序的且大表的分桶数量是小表分桶数量的整数倍。

此时就可以以分桶为单位为每个Map分配任务了Map端就无需再缓存小表的全表数据了而只需缓存其所需的分桶。

6.2

hive.auto.convert.join.noconditionaltask.size

6.2.3

hive.optimize.bucketmapjoin.sortedmergetrue;

–使用自动转换SMB

hive.auto.convert.sortmerge.jointrue;

6.3.2

上述SQL语句共有两张表一次join操作故优化前的执行计划应包含一个Common

3**优化思路**

表名大小dwd_trade_order_detail_inc162900000000约160gdim_user_zip12320000000

约12g

首先需要依据源表创建两个的有序的分桶表dwd_trade_order_detail_inc建议分36个bucketdim_user_zip建议分6个bucket,注意**分桶个数的倍数关系以及分桶字段和排序字段**。

–****订单明细表

dwd_trade_order_detail_inc_bucketed;

create

dwd_trade_order_detail_inc_bucketed(

string,

dwd_trade_order_detail_inc_bucketed

select

hive.optimize.bucketmapjoin.sortedmergetrue;

–使用自动转换SMB

hive.auto.convert.sortmerge.jointrue;

最后在重写SQL语句如下

dwd_trade_order_detail_inc_bucketed

join

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

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

7.2

hive.map.aggr.hash.percentmemory0.5;

2**启用skew

其原理是启动两个MR任务第一个MR按照随机数分区将数据分散发送到Reduce完成部分聚合第二个MR按照分组字段分区完成最终聚合。

相关参数如下

hive.auto.convert.join.noconditionaltask.size

使用map

对于一个分布式的计算任务而言设置一个合适的并行度十分重要。

在Hive中无论其计算引擎是什么所有的计算任务都可分为Map阶段和Reduce阶段。

所以并行度的调整也可从上述两个方面进行调整。

8.2

Map端的并行度也就是Map的个数。

是由输入文件的切片数决定的。

一般情况下Map端的并行度无需手动调整。

Map端的并行度相关参数如下

set

hive.input.formatorg.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

set

mapreduce.input.fileinputformat.split.maxsize256000000;

8.3

Reduce端的并行度相对来说更需要关注。

默认情况下Hive会根据Reduce端输入数据的大小估算一个Reduce并行度。

但是在某些情况下其估计值不一定是最合适的故需要人为调整其并行度。

Reduce并行度相关参数如下

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上游的Operator的Statistics统计信息中获取的。

为保证Hive能获得准确的统计信息需配置如下参数

set

hive.stats.column.autogathertrue;

–计算Reduce并行度时从上游Operator统计信息获得输入数据量

set

hive.stats.fetch.column.statstrue;

第9章小文件合并优化

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

9.2

合并Map端输入的小文件是指将多个小文件划分到一个切片中进而由一个Map

相关参数为

hive.input.formatorg.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

9.3

合并Reduce端输出的小文件是指将多个小文件合并成大文件。

目的是减少HDFS小文件数量。

相关参数为

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

2.https://cwiki.apache.org/confluence/display/Hive/HiveonSpark%3AGettingStarted

Spark性能调优完整版