Spark 调优 — 资源调优&参数调优

• spark-submit 参数

具体如下：

spark-submit \
--queue ${1} \ # 指定资源队列，否则默认default
--master yarn
--deploy-mode ${2} \ # client or cluster: 选择客户端模式或者集群模式提交：决定driver 是在本地还是集群内节点：cluster模式能够降低本地压力，发挥集群服务器优势
--num-executors ${3} \ # 启动的executor进程数量：50-100个比较合适，具体看可用资源
--executor-cores ${4} \ # 每个executor进程分配的cpu cores，决定了可并发执行的task数量（一个cpu只能同时运行一个task）: 设置4-6个
--executor-memory ${5} \ # 每个executor进行可用的内存, 直接决定Spark 作业性能
--driver-cores ${6} \ # driver端cpu cores，一般2个，取决于任务是否复杂，driver端压力大的可稍微调大点
--driver-memory ${7} \ # driver端可用内存，1G左右差不多，若经常需要collect把数据拉到driver端处理，需要根据数据量调大内存
--class ${8} ${9} # 依次是执行的class名称，执行的jar包路径

• Kryo序列化

Spark为使用者提供两种序列化方式：
Java Serialization: 默认的序列化方式，也就是ObjectOutputStream/ObjectInputStream API来进行序列化和反序列化。
Kryo Serialization: 相较于 Java Serialization 的方式，速度更快，空间占用更小，但并不支持所有的序列化格式，同时使用的时候需要注册class。spark-sql中默认使用的是kyro的序列化方式。
conf.set(“spark.serializer”, “org.apache.spark.serializer.KryoSerializer”) ，该参数会同时作用于机器之间数据的shuffle操作以及序列化rdd到磁盘，内存
如遇关于注册的报错，则在代码中注册即可：

 sparkConf.registerKryoClasses(
    Array(classOf[scala.collection.mutable.WrappedArray.ofRef[_]],
    classOf[MyClass]))
val conf = new SparkConf().setMaster("local[2]").setAppName("KyroTest")
      conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      conf.registerKryoClasses(Array(classOf[Info]))
  val sc = new SparkContext(conf)

• Application Properties 应用基本属性

spark.driver.cores  

driver端分配的核数，默认为1，thriftserver是启动thriftserver服务的机器，资源充足的话可以尽量给多。

spark.driver.memory

driver端分配的内存数，默认为1g，同上。

spark.executor.memory

每个executor分配的内存数，默认1g，会受到yarn CDH的限制，和memoryOverhead相加 不能超过总内存限制。

spark.driver.maxResultSize

driver端接收的最大结果大小，默认1GB，最小1MB，设置0为无限。

这个参数不建议设置的太大，如果要做数据可视化，更应该控制在20-30MB以内。

过大会导致OOM。

spark.extraListeners

默认none，随着SparkContext被创建而创建，用于监听单参数、无参数构造函数的创建，并抛出异常。

• Shuffle Behavior

spark.reducer.maxSizeInFlight

默认48m。从每个reduce任务同时拉取的最大map数，每个reduce都会在完成任务后，需要一个堆外内存的缓冲区来存放结果，如果没有充裕的内存就尽可能把这个调小一点。。相反，堆外内存充裕，调大些就能节省gc时间。

spark.reducer.maxBlocksInFlightPerAddress

限制了每个主机每次reduce可以被多少台远程主机拉取文件块，调低这个参数可以有效减轻node manager的负载。（默认值Int.MaxValue）

spark.reducer.maxReqsInFlight

限制远程机器拉取本机器文件块的请求数，随着集群增大，需要对此做出限制。否则可能会使本机负载过大而挂掉。。（默认值为Int.MaxValue）

spark.reducer.maxReqSizeShuffleToMem

shuffle请求的文件块大小 超过这个参数值，就会被强行落盘，防止一大堆并发请求把内存占满。（默认Long.MaxValue）

spark.shuffle.compress

是否压缩map输出文件，默认压缩 true

spark.shuffle.spill.compress

shuffle过程中溢出的文件是否压缩，默认true，使用spark.io.compression.codec压缩。

spark.shuffle.file.buffer

在内存输出流中 每个shuffle文件占用内存大小，适当提高 可以减少磁盘读写 io次数，初始值为32k

spark.shuffle.memoryFraction

该参数代表了Executor内存中，分配给shuffle read task进行聚合操作的内存比例，默认是20%。

cache少且内存充足时，可以调大该参数，给shuffle read的聚合操作更多内存，以避免由于内存不足导致聚合过程中频繁读写磁盘。

spark.shuffle.manager

当ShuffleManager为SortShuffleManager时，如果shuffle read task的数量小于这个阈值（默认是200），则shuffle write过程中不会进行排序操作，而是直接按照未经优化的HashShuffleManager的方式去写数据，但是最后会将每个task产生的所有临时磁盘文件都合并成一个文件，并会创建单独的索引文件。

当使用SortShuffleManager时，如果的确不需要排序操作，那么建议将这个参数调大一些，大于shuffle read task的数量。那么此时就会自动启用bypass机制，map-side就不会进行排序了，减少了排序的性能开销。但是这种方式下，依然会产生大量的磁盘文件，因此shuffle write性能有待提高。

spark.shuffle.consolidateFiles

如果使用HashShuffleManager，该参数有效。如果设置为true，那么就会开启consolidate机制，会大幅度合并shuffle write的输出文件，对于shuffle read task数量特别多的情况下，这种方法可以极大地减少磁盘IO开销，提升性能。

如果的确不需要SortShuffleManager的排序机制，那么除了使用bypass机制，还可以尝试将spark.shuffle.manager参数手动指定为hash，使用HashShuffleManager，同时开启consolidate机制。

spark.shuffle.io.maxRetries

shuffle read task从shuffle write task所在节点拉取属于自己的数据时，如果因为网络异常导致拉取失败，是会自动进行重试的。该参数就代表了可以重试的最大次数。如果在指定次数之内拉取还是没有成功，就可能会导致作业执行失败。

对于那些包含了特别耗时的shuffle操作的作业，建议增加重试最大次数（比如60次），以避免由于JVM的full gc或者网络不稳定等因素导致的数据拉取失败。在实践中发现，对于针对超大数据量（数十亿~上百亿）的shuffle过程，调节该参数可以大幅度提升稳定性。

spark.shuffle.io.retryWait

同上，默认5s，建议加大间隔时长（比如60s），以增加shuffle操作的稳定性。

spark.io.encryption.enabled + spark.io.encryption.keySizeBits + spark.io.encryption.******.algorithm

io加密，默认关闭

• Compression and Serialization

spark.broadcast.compress

广播变量前是否会先进行压缩。默认true （spark.io.compression.codec）

spark.io.compression.codec

压缩RDD数据、日志、shuffle输出等的压缩格式 默认lz4

spark.io.compression.lz4.blockSize

使用lz4压缩时，每个数据块大小 默认32k

spark.rdd.compress

rdd是否压缩 默认false，节省memory_cache大量内存 消耗更多的cpu资源（时间）。

spark.serializer.objectStreamReset

当使用JavaSerializer序列化时，会缓存对象防止写多余的数据，但这些对象就不会被gc，可以输入reset 清空缓存。默认缓存100个对象，修改成-1则不缓存任何对象。

• Memory Management

spark.memory.fraction

执行内存和缓存内存（堆）占jvm总内存的比例，剩余的部分是spark留给用户存储内部源数据、数据结构、异常大的结果数据。

默认值0.6，调小会导致频繁gc，调大容易造成oom。

spark.memory.storageFraction

用于存储的内存在堆中的占比，默认0.5。调大会导致执行内存过小，执行数据落盘，影响效率；调小会导致缓存内存不够，缓存到磁盘上去，影响效率。

值得一提的是在spark中，执行内存和缓存内存公用java堆，当执行内存没有使用时，会动态分配给缓存内存使用，反之也是这样。如果执行内存不够用，可以将存储内存释放移动到磁盘上（最多释放不能超过本参数划分的比例），但存储内存不能把执行内存抢走。

spark.memory.offHeap.enabled

是否允许使用堆外内存来进行某些操作。默认false

spark.memory.offHeap.size

允许使用进行操作的堆外内存的大小，单位bytes 默认0

spark.memory.useLegacyModes

默认false，不开启，在spark1.5之后就被废弃了，下面三个参数也是。。就不做更多的介绍了：

spark.shuffle.memoryFraction

spark.storage.memoryFraction

spark.storage.unrollFraction

spark.storage.replication.proactive

针对失败的executor，主动去cache 有关的RDD中的数据。默认false

spark.cleaner.periodicGC.interval

控制触发gc的频率，默认30min

spark.cleaner.referenceTracking

是否进行context cleaning，默认true

spark.cleaner.referenceTracking.blocking

清理线程是否应该阻止清理任务，默认true

spark.cleaner.referenceTracking.blocking.shuffle

清理线程是否应该阻止shuffle的清理任务，默认false

spark.cleaner.referenceTracking.cleanCheckpoints

清理线程是否应该清理依赖超出范围的检查点文件（checkpoint files不知道怎么翻译。。）默认false

• Networking

spark.rpc.message.maxSize

executors和driver间消息传输、map输出的大小，默认128M。map多可以考虑增加。

spark.driver.blockManager.port和spark.driver.bindAddress

driver端绑定监听block manager的地址与端口。

spark.driver.host和spark.driver.port

driver端的ip和端口。

spark.network.timeout

网络交互超时时间，默认120s。如果

spark.core.connection.ack.wait.timeout

spark.storage.blockManagerSlaveTimeoutMs

spark.shuffle.io.connectionTimeout

spark.rpc.askTimeout orspark.rpc.lookupTimeout

没有设置，那么就以此参数为准。

spark.port.maxRetries

设定了一个端口后，在放弃之前的最大重试次数，默认16。 会有一个预重试机制，每次会尝试前一次尝试的端口号+1的端口。如 设定了端口为8000，则最终会尝试8000~(8000+16)范围的端口。

spark.rpc.numRetries

rpc任务在放弃之前的重试次数，默认3，即rpc task最多会执行3次。

spark.rpc.retry.wait

重试间隔，默认3s

spark.rpc.askTimeout

rpc任务超时时间，默认spark.network.timeout

spark.rpc.lookupTimeout

rpc任务查找时长

• Scheduling

spark.scheduler.maxRegisteredResourcesWaitingTime

在执行前最大等待申请资源的时间，默认30s。

spark.scheduler.minRegisteredResourcesRatio

实际注册的资源数占预期需要的资源数的比例，默认0.8

spark.scheduler.mode

调度模式，默认FIFO 先进队列先调度，可以选择FAIR。

spark.scheduler.revive.interval

work回复重启的时间间隔，默认1s

spark.scheduler.listenerbus.eventqueue.capacity

spark事件监听队列容量，默认10000，必须为正值，增加可能会消耗更多内存

spark.blacklist.enabled

是否列入黑名单，默认false。如果设成true，当一个executor失败好几次时，会被列入黑名单，防止后续task派发到这个executor。可以进一步调节spark.blacklist以下相关的参数：

（均为测试参数 Experimental）

spark.blacklist.timeout

spark.blacklist.task.maxTaskAttemptsPerExecutor

spark.blacklist.task.maxTaskAttemptsPerNode

spark.blacklist.stage.maxFailedTasksPerExecutor

spark.blacklist.application.maxFailedExecutorsPerNode

spark.blacklist.killBlacklistedExecutors

spark.blacklist.application.fetchFailure.enabled

spark.speculation

推测，如果有task执行的慢了，就会重新执行它。默认false，

详细相关配置如下：

spark.speculation.interval

检查task快慢的频率，推测间隔，默认100ms。

spark.speculation.multiplier

推测比均值慢几次算是task执行过慢，默认1.5

spark.speculation.quantile

在某个stage，完成度必须达到该参数的比例，才能被推测，默认0.75

spark.task.cpus

每个task分配的cpu数，默认1

spark.task.maxFailures

在放弃这个job前允许的最大失败次数，重试次数为该参数-1，默认4

spark.task.reaper.enabled

赋予spark监控有权限去kill那些失效的task，默认false

(原先有 job失败了但一直显示有task在running，总算找到这个参数了)

其他进阶的配置如下：

spark.task.reaper.pollingInterval

轮询被kill掉的task的时间间隔，如果还在running，就会打warn日志，默认10s。

spark.task.reaper.threadDump

线程回收是是否产生日志，默认true。

spark.task.reaper.killTimeout

当一个被kill的task过了多久还在running，就会把那个executor给kill掉，默认-1。

spark.stage.maxConsecutiveAttempts

在终止前，一个stage连续尝试次数，默认4。

• Dynamic Allocation 动态分配（Spark2.0之后）
  spark.dynamicAllocation.enabled
  是否开启动态资源配置，根据工作负载来衡量是否应该增加或减少executor，默认false

  因涉及资源配置，需要在资源申请前设置好，而不能在程序代码里再重新设置：即需要在spark-submit处就设置好，非SparkSession.builder().config(“参数”,“参数值”)设置

以下相关参数：

spark.dynamicAllocation.minExecutors

动态分配最小executor个数，在启动时就申请好的，默认0

spark.dynamicAllocation.maxExecutors

动态分配最大executor个数，默认infinity

spark.dynamicAllocation.initialExecutors

动态分配初始executor个数默认值=spark.dynamicAllocation.minExecutors

spark.dynamicAllocation.executorIdleTimeout

当某个executor空闲超过这个设定值，就会被kill，默认60s

spark.dynamicAllocation.cachedExecutorIdleTimeout

当某个缓存数据的executor空闲时间超过这个设定值，就会被kill，默认infinity

spark.dynamicAllocation.schedulerBacklogTimeout

任务队列非空，资源不够，申请executor的时间间隔，默认1s

spark.dynamicAllocation.sustainedSchedulerBacklogTimeout

同schedulerBacklogTimeout，是申请了新executor之后继续申请的间隔，默认=schedulerBacklogTimeout

• SparkSteaming

spark.streaming.backpressure.enabled

反压，默认false，详细了解请移步https://blog.csdn.net/zyzzxycj/article/details/82385982

spark.streaming.backpressure.initialRate

当开启反压机制时，每个receiver在最初，能从每个batch接收的最大速率，默认not set。

spark.streaming.blockInterval

对于某个batch，receiver将其转化成数据的时间间隔，默认200ms，不建议调到小于50ms，不然处理数据的时间就会比申请线程的时间还小。

因此可以计算出Tasks=BatchInterval / BlockInterval

spark.streaming.receiver.maxRate

每个receiver接收的最大速率(每秒接收记录数)，如果是<=0 则为无限，默认not set。

spark.streaming.receiver.writeAheadLog.enable

receiver日志，便于失败时的恢复，默认false

spark.streaming.unpersist

自动将spark streamin*生的、持久化的数据给清理掉，默认true，自动清理内存垃圾。

spark.streaming.stopGracefullyOnShutdown

优雅的关闭spark streaming jvm，默认false 

spark.streaming.ui.retainedBatches

spark streaming 日志接口在gc时保留的batch个数，默认1000