1，Spark参数调优

时间：2022-02-04 23:52:54

Spark调优

目录

一、代码规范

  • 调优顺序：spark任务的调优顺序依次是代码规范、资源参数(并行度)、数据倾斜、shuffle调优、业务层面；

1.1 避免创建重复RDD

  • 对于新手，或者一些较为复杂的spark任务，可能会忘记之前对于某一份数据已经创建过一个RDD，而重复创建，造成不必要的计算；

1.2 尽量复用同一个RDD

  • 下游需要使用key-value类型和key类型的两个RDD，这两个RDD的数据完全相同，只是格式不同，那么就只需要创建key-value这一个RDD就行，而使用key类型的RDD直接复用key-value类型的RDD就行了；因此，对于需要使用数据相同，格式不同的数据源时，最好复用字段较多的RDD；

1.3 多次使用的RDD要持久化

  • 当一个RDD被使用了多次，比如上面的复用同一个RDD，那么这个RDD就要做持久化，否则这个RDD就会被计算多次；例如，a = rdd1.map(); b = rdd1.map(); 那么就需要对 rdd1做持久化rdd1.persist()，否则rdd1就会被计算两次；

1.4 使用高性能算子

  • 用reduceByKey替代groupByKey求聚合：前者是map-side预聚合算子，会在map端预聚合，类似于Combiner；
  • 用combineByKey代替groupByKey求topN：前者可以自定义分区内合并和分区间合并的计算逻辑，也是预聚合；
  • mapPartition替代map：一次调用处理一个分区的数据，对于需要在map中创建很多重复对象的场景，最好使用mapPartition，同时注意OOM问题；
  • foreachPartition替代foreach：道理同mapPartition一样；在需要将rdd的数据写入MySQL时，后者是一条一条数据插入，并且每条数据都会创建一次数据库连接；而前者则是一个分区操作一次，性能有很高的提升；

1.5 好习惯

  • 广播大变量：当需要在算子中使用大变量（1g以内）时，最好将大变量广播到Executor中，例如：rdd1.filter(x=>slant.contains(x))，如果slant在20M~1G之间，就可以将slant广播；

  • filter后coalesce：由于filter后，各个分区中的数据不再均衡，使用coalesce再平衡一下分区数据；

  • 优化数据结构：对于算子中的数据结构，能用数组就不要用集合类型，最好使用字符串代替对象，用基本类型代替字符串；

  • 使用Kryo序列化：spark中的三个场景会涉及到序列化，算子中使用外部变量、将自定义对象作为RDD中的类型、可序列化的持久化策略（如MEMORY_ONLY_SER），使用kryo的性能会高很多；使用Kryo序列化时，最好注册所有的自定义类；conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))；

  • persist后unpersist：unpersist是立即释放缓存，对复用的RDD使用persist缓存后，需要使用行动算子提交job后，才会真正的缓存，然后再使用unpersist释放缓存；所以当persist缓存的RDD不会再使用时，最好是手动unpersist释放缓存；

二、参数调优

资源参数

1.1 --num-executors 100

  • 参数解释：任务可以申请的Excutor最大数量，并不是一次性分配100个Excutor；Excutor数量会在任务的运行过程中动态调整，有 job处于pending状态则申请Excutor，一个Excutor空闲时间过长则将其移除；Excutor的数量决定了任务的并行度；

  • 申请Excutor：当有任务处于pending状态（积压）超过一定时间，就认为资源不足，需要申请Excutor；

    何时申请：当pending积压的任务超过spark.dynamicAllocation.schedulerBacklogTimeout（1秒）就申请
    
申请多少：申请数量 = 正在运行和pending的任务数量 * spark.dynamicAllocation.executorAllocationRatio（1）/ 并行度

  • 移除Excutor：

    spark.dynamicAllocation.enabled（false）决定是否使用资源动态分配；必须开启外部shuffle；
    
spark.dynamicAllocation.executorIdleTimeout （60s）空闲60s就会被回收(并且没有缓存)；

  • 决定任务的并行度：executor的数量就是工作节点的数量，直接决定了任务的并行度；准确的说是由executor*core决定的；这只是物理上提供的最大并行度，而任务实际的并行度还是由程序中设置的并行度决定，也就是RDD的分区数；

1.2 --executor-memory 5g

  • 参数解释：每个executor的内存大小；对于spark调优和OOM异常，通常都是对executor的内存做调整，spark内存模型也是指executor的内存分配，所以executor的内存管理是非常重要的；
  • 内存分配：该参数是总的内存分配，而在任务运行中，会根据spark内存模型对这个总内存再次细分；在实际生产中，通常需要根据程序中使用的缓存内存和计算内存，来划分不同的比例，从而合理的利用内存，避免OOM，提高性能；

1.3 --executor-cores 4

  • 参数解释：每个executor的核数；是每个executor内部的并行度，即一个executor中可同时执行的task数量；
  • 并行度：core的数量决定了一个executor同时执行的task数量，如果task数量越多，则意味着占用的executor内存也越多；所以，在executor内存固定的情况下，可以通过增加executor数量，减少core数量，使任务总并行度不变的前提下，降低OOM风险；如果任务需要广播大变量，可以增大core数，使更多的task共用广播变量；

1.4 --driver-memory

  • 参数解释：driver端的内存大小；如果要collect大量数据到driver端，或者要广播大变量时，就需要调大driver端的内存；一般给个3G、4G就够了；

内存参数

spark.storage.memoryFraction、spark.shuffle.memoryFraction（spark1.6之前静态内存管理）

  • 参数解释：在spark1.6之前，使用的是静态内存管理，而这两个参数就是用来决定缓存内存和执行内存大小的；在spark1.6及之后，采用的是统一内存管理（也叫动态内存管理），这两个参数就废弃了（但也可以让它生效）

spark.memory.fraction（spark1.6及之后，统一内存管理）

  • 参数解释：spark1.6及之后采用的是统一内存管理，也叫动态内存管理，顾名思义，就是缓存内存和执行内存统一管理，并且是动态的；首先解释“统一”：spark.memory.fraction是堆内内存中用于执行、shuffle、缓存的内存比例；这个值越低，则执行时溢出到磁盘更频繁、同时缓存被逐出内存也更频繁；一般使用默认值就好了，spark2.2默认是0.6，那么剩下的0.4就是用于存储用户的数据结构（比如map算子中定义的中间数据）以及spark内部的元数据；

spark.memory.storageFraction

  • 参数解释：存储内存不会被逐出内存的总量，这个是基于spark.memory.fraction的占比；这个值越高，则执行、shuffle的内存就越少，从而溢写到磁盘就越频繁；一般使用默认值就好了，spark2.2默认是0.5；

spark.kryoserializer.buffer.max

  • 参数解释：kryo序列化时使用的缓存大小；如果collect大量数据到driver端，可能会抛buffer limit exceeded异常，这个时候就要调大该参数；默认是64m，挂了就设置为1024m；如果序列化的一个对象很大，那么就需要增大改参数的值spark.kryoserializer.buffer(默认64k)；

dfs.client.block.write.locateFollowingBlock.retries

  • 参数解释：写入块后尝试关闭的次数；Unable to close file because the last block does not have enough number of replicas异常的原因；2.7.4已修复；默认是5，挂了就设置为6；

spark.driver.maxResultSize

  • 参数解释：一次collect到driver端的最大内存大小，Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)异常时需要调大该值；默认1g，挂了就设置为2g，0表示不限制；

shuffle参数

spark.shuffle.file.buffer

  • 参数解释：shuffle write时，会先写到BufferedOutputStream缓冲区中，然后再溢写到磁盘；该参数就是缓存区大小，默认32k，建议设置为64k；

spark.shuffle.spill.batchSize

  • 参数解释：shuffle在spill溢写过程中需要将数据序列化和反序列化，这个是一个批次处理的条数；默认是10000，可以调大该值，2万5万都可以；

spark.shuffle.io.maxRetries

  • 参数解释：shuffle read拉取数据时，由于网络异常或者gc导致拉取失败，会自动重试，改参数就是配置重试次数，在数据量达到十亿、百亿级别的时候，最好调大该参数以增加稳定性；默认是3次，建议设置为10到20；

spark.shuffle.io.retryWait

  • 参数解释：该参数是 spark.shuffle.io.maxRetries的重试间隔，默认是5s，建议设置为20s；

spark.reducer.maxSizeInFlight

  • 参数解释：shuffle read拉取数据时的缓存区大小，也就是一次拉取的数据大小；注意是从5个节点拉取48M的数据，而不是从一个节点获取48M；默认48m，建议设置为96m；
  • 原理解释：从远程节点拉取数据时，是并行的从发送5个请求，每个请求拉取的最大长度是 48M / 5，但是拉取时都是以block为最小单位的，所以实际获取的有可能会大于这个值；

spark.reducer.maxReqsInFlight

  • 参数解释：shuffle read时，一个task的一个批次同时发送的请求数量；默认是 Int的最大值；
  • 原理解释：构造远程请求时，单个请求大小限制是 48M / 5，而在一次拉取远程block数据时，是按批次拉取，一个批次的大小限制是 48M，所以理想情况下一个批次会发送5个请求；但如果block的分布不均匀，导致一个请求的请求大小远小于 48M / 5 (例如1M)，而一个批次的大小限制是48M，所以这个批次就会发送48个请求；当节点数较多时，一个task的一个批次可能会发送非常多的请求，导致某些节点的入站连接数过多，从而导致失败；

spark.reducer.maxReqSizeShuffleToMem

  • 参数解释：shuffle read时，从远程拉取block如果大于这个值就会强行落盘，默认是Long的最大值，建议小于2G，一般设为200M，spark2.2开始生效；（spark2.3开始换成了这个参数spark.maxRemoteBlockSizeFetchToMem）；shuffle read这个部分的参数在spark的版本更新中变化较大，所以在优化时一定要根据集群的spark版本设置对应的参数；
  • 原理解释：一次拉取请求中，如果要拉取的数据比较大，内存放不下，就直接落盘；对于数据倾斜比较严重的任务，有可能一个block非常大，而没有足够的内存存放时就会OOM，所以最好限制该参数的大小；还有一个原因就是 netty的最大限制是2G，所以大于2G肯定会报错；spark2.4该参数的默认值是：Int的最大值-512 (2G，减512用来存储元数据)；spark3.0的最大值也是2G，并且给了默认值200M；

spark.reducer.maxBlocksInFlightPerAddress

  • 参数解释：shuffle read时，一个节点同时被拉取的最大block数，如果太多可能会导致executor服务或nodemanager崩溃；默认Int的最大值；（spark2.2.1开始支持）；

  • 原理解释：shuffle read时每个task都会从shuffle write所在的节点拉取自己的block数据，如果一个shuffle write的executor运行了9个task，就会write9个data文件；如果shuffle read有1000核，那么同时运行1000个task，每个task要到shuffle write所在的executor获取9个block，极端情况下一个shuffle write的executor会被请求9000次；当节点数非常多时，一个shuffle write的executor会同时被很多节点拉取block，从而导致失败；

文件相关

spark.sql.files.maxPartitionBytes

  • 参数解释：sparksql读取文件时，每个分区的最大文件大小，这个参数决定了读文件时的并行度；默认128M；例如一个300M的text文件，按128M划分为3个切片，所以SparkSQL读取时最少有3个分区；
  • 原理解释：sparksql读取文件的并行度=max(spark默认并行度，切片数量(文件大小/ 该参数))；这里要注意压缩文件是否可分割；但是要注意，对于parquet格式，一个切片对应一个row group；

spark.sql.parquet.compression.codec

  • 参数解释：parquet格式的压缩方式，默认是snappy，可以选择gzip、lzo、或者uncompressed不压缩；

spark.io.compression.codec

  • 参数解释：spark中rdd分区、广播变量、shuffle输出的压缩格式，spark2.2默认是lz4；

spark.serializer

  • 参数解释：spark序列化的实现，这里的序列化是针对shuffle、广播和rdd cache的序列化方式；默认使用java的序列化方式org.apache.spark.serializer.JavaSerializer性能比较低，所以一般都使用org.apache.spark.serializer.KryoSerializer ，使用Kryo序列化时最好注册十分需要空间的类型，可以节省很多空间；spark task的序列化由参数spark.closure.serializer配置，目前只支持JavaSerializer；

spark.sql.hive.convertMetastoreParquet

  • 参数解释：是否采用spark自己的Serde来解析Parquet文件；Spark SQL为了更好的性能，在读取hive metastore创建的parquet文件时，会采用自己Parquet Serde，而不是采用hive的Parquet Serde来序列化和反序列化，这在处理null值和decimal精度时会有问题；默认为true，设为false即可(会采用与hive相同的Serde)；

spark.sql.parquet.writeLegacyFormat

  • 参数解释：是否使用遗留的(hive的方式)format来写Parquet文件；由于decimal精度问题，hive读取spark创建的Parquet文件会报错；所以这里的spark采用与hive相同的writeFormat来写Parquet文件，这样hive在读取时就不会报错；并且上下游表的精度最好一致，例如a表的字段精度为decimal(10,2)，b表也最好是decimal(10,2)；
  • 原理解释：在hive中decimal类型是固定的用int32来表示，而标准的parquet规范约定，根据精度的不同会采用int32和int64来存储，而spark就是采用的标准的parquet格式；所以对于精度不同decimal的，底层的存储类型有变化；所以使用spark存储的parquet文件，在使用hive读取时报错；将spark.sql.parquet.writeLegacyFormat(默认false)配置设为true，即采用与hive相同的format类来读写parquet文件；

参考文章

相关文章