以下参数中有sql字眼的一般只有spark-sql模块生效,例外的时候会另行说明,此外由于总结这些参数是在不同时间段,当时使用的spark版本也不一样,因此要注意是否有效,如果本博主已经试过的会直接说明。
1、任务使用资源限制,基本参数,注意,这些资源配置有spark前缀是因为他们是标准的conf配置,也就是submit脚本,你调用–conf参数写的,和–driver.memory这种属于不同的优先级,–driver.memory这种优先级比它高,对于spark来讲,数据量和计算量是两个不同的概念,随着执行计划的不同,往往计算量是数据量的n倍,因此要使用的计算资源也是比较多的,因为除了计算本身,还有伴随计算产生的其他消耗,因此往往1G的数据要付出3G的计算资源,甚至更多,在具体计算的时候使用多少资源,就需要经验了,这是只可意会无法言传的东西
spark.driver.memory=20G #applicationmaster启动的driver进程占用内存
spark.driver.cores=4 #applicationmaster启动的driver进程占用核数
spark.executor.cores=4 #容器占用核数
spark.executor.memory=40G #容器占用的内存数
spark.num.executors=10 #任务用到的总容器数,如果你开启了容器伸缩哪这个配置无效了
2、限制sql任务运行时生成的分区和拉去文件总大小的上线
spark.sql.watchdog.maxPartitions=2000
spark.sql.watchdog.maxFileSize=3t
3、文件聚合,文件聚合的阈值会参照分区大小决定,就是说去设置AQE
spark.sql.optimizer.insertRepartitionBeforeWriteIfNoShuffle.enabled=true #是否对不发生shuffer的stage做聚合
spark.sql.optimizer.insertRepartitionBeforeWrite.enabled=true #是否在写入文件之间聚合
spark.sql.optimizer.finalStageConfigIsolation.enabled=true #最后任务的最后阶段文件聚合,会有一个落盘前聚合的执行计划
4、最后阶段消耗资源多少,可以配合压缩和自适应分区的相关配置用来优化数据落盘的大小
spark.sql.finalWriteStage.executorMemory=10g
spark.sql.finalWriteStage.executorCores=2
5、文件压缩
mapreduce.output.fileoutputformat.compress=true #是否对任务的输出数据压缩
mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.GzipCodec #用到的压缩类
mapreduce.map.output.compress=true #是否对map阶段的输出进行压缩
mapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.GzipCodec #同上
hive.exec.compress.output=true #hive端数据压缩,⭐️⭐️3.x的spark之后共用上面map的压缩策略,但是2.x的版本有一个mapred.map.output.compress.codec,使用的时候注意版本
spark.sql.parquet.compression.codec=snappy #如果表数据存储类型是parquet,那就要另行制定压缩方式,默认是snappy,可以选择gzip、lzo、或者uncompressed不压缩
6、sql任务shuffer阶段,read数据时的分区数,用来提升shuffer读取效率
spark.sql.shuffle.partitions=200
7、sql任务自适应分区数查询(AQE),注意同时配置了AQE的合并和倾斜时,会先合并,再调整倾斜,设计到的分区大小,建议50~300M
spark.sql.adaptive.enabled=true # 开启aqe
spark.sql.adaptive.coalescePartitions.enabled=true # 自动合并分区
spark.sql.adaptive.coalescePartitions.initialPartitionNum=100 # 初始的分区数。默认为spark.sql.shuffle.partitions的值
spark.sql.adaptive.coalescePartitions.minPartitionNum=1 # 最小的分区数。默认为spark.sql.shuffle.partitions的值,一般不另行配置
spark.sql.adaptive.coalescePartitions.maxPartitionNum=1 # 最大的分区数。一般不另行配置因为有上面的最大任务限制
spark.sql.adaptive.advisoryPartitionSizeInBytes=128M # 每个分区建议大小(默认单位字节) ⭐️⭐️注意这个配置在3.x之后支持
spark.sql.adaptive.shuffle.targetPostShuffleInputSize=128M #设置shuffer阶段后下一阶段任务输入预期的数据大小,一般不另行配置
spark.sql.adaptive.localShuffleReader.enabled=true #允许在自适应时采用本地进程优化shuffer,分险是如果报错,这部分日志无法聚合到yarn
spark.sql.adaptive.skewJoin.enabled=true # 开启数据倾斜时join检测
spark.sql.adaptive.skewJoin.skewedPartitionFactor=5 # 默认5,当某个分区大小大于所有分区的大小中间值5倍,就打散数据
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256M #通过直接指定分区大小的阈值来决定是否打散分区, 默认256M,和上面的参数一起生效,用来因对不同的情况
spark.sql.mergesmallfilesize=256M #分区建议大小 ⭐️⭐️注意这个配置在3.x之后废弃
对于AQE的分区数,一定要知道不是说你设置多少,web页面上就能直观的体现出来多少,执行计划上是以task的形式展示执行计划的,而task和分区是两个东西,因此你要有那种感觉,来调整分区数,这种感觉只能靠经验去喂,就和神枪手一样。
而且AQE的
coalescePartitions
对数据集倾斜的作用很明显,但是对join是发生的热点key倾斜就不太有力了,所以如果有join倾斜一定要开启skewJoin
8、任务容器的动态伸缩
spark.dynamicAllocation.enabled=true # 开启动态收缩容器资源默认false
spark.dynamicAllocation.shuffleTracking.enabled=true # shuffle动态跟踪,默认true
spark.dynamicAllocation.initialExecutors # 初始化申请资源
spark.dynamicAllocation.maxExecutors # 最大容器个数
spark.dynamicAllocation.minExecutors # 最小容器个数
spark.dynamicAllocation.executorAllocationRatio # 这个用来设置动态容器资源模式下,容器自身的占用核数为多少,默认为1,一般不另行配置
9、是否对分区做动态裁剪,默认true,这个配置一般不关,它目的就是优化执行,开启后你可以在spark的任务web界面看到有的执行计划就被skip了,当然skip不全是因为它,容器的动态伸缩和自适应分区数也会造成。
spark.sql.optimizer.dynamicPartitionPruning.enabled
10、spark-sql提供了
Hint
,需要你去查看官方文档
https://spark.apache.org/docs/3.5.2/sql-ref-syntax-qry-select-hints.html#content
,看的时候注意你用的版本,这个就是在写sql的时候加入建议的执行计划,比如当你希望sql执行的时候直接指定希望的分区数,你可以写成如下的格式,但是这种方式其实就是嫌少了用户使用时的代码量,一般用的不多
SELECT/*+ COALESCE(3) */*FROM t;
11、core任务设置并行度,这个配置一般不直接在任务外配置,有需要的话调用算子的parallelism方法了
spark.default.parallelism=10
12、存储内存占用比例,这个配置越大留给shuffer和计算本身的内存就越少,反之越小跑任务的时候数据暂时落盘的次数就越频繁,默认值0.5
spark.memory.storageFraction=0.5
对于落盘的阈值,在整个spark中有个
spark.reducer.maxReqSizeShuffleToMem
参数,用来直接用数据大小来控制落盘时机,但是该参数变动很频繁,不同版本名称也不一样,所以一般不用
13、kryo序列化时使用的缓存大小,这个配置是当你有需要让driver调用collect收集大量数据到driver端,可能会抛buffer limit exceeded异常,这个时候就要调大该参数
spark.kryoserializer.buffer=64k
14、第13点设置的是缓存大小,这个配置设置的是driver收集数据使用的内存资源最大是多少,默认1g,0表示不限制
spark.driver.maxResultSize=1g
15、下一个数据块定位次数,在数据落盘的时候如果网络延迟等极端原因会导致driver定位数据块写入位置时,收不到任何datanode的回馈,这个时候可以尝试调大这个值,一般不会遇到,博主只遇到过一次,出现问题的时候会抛出Unable to close file because the last block does not have enough number of replicas异常,对应的bug在spark2.7.4已修复,这个配置的默认值是5,挂了就设置为6
dfs.client.block.write.locateFollowingBlock.retries=5
16、shuffle write时,会先写到BufferedOutputStream缓冲区中,然后再写到磁盘,该参数就是缓存区大小,默认32k,建议设置为64k,这个配置是数据量不较大的时候,减少一些系列化次数,和让小文件聚合异曲同工,设置的时候注意要和17平衡
spark.shuffle.file.buffer=32k
17、shuffle溢写磁盘过程中需要将数据序列化和反序列化,这个参数是一个批次处理的条数,默认是10000,需要的话调大该值,2万5万都可以,但是一定要成比例的设置16的配置值
spark.shuffle.spill.batchSize=10000
18、shuffle read拉取数据时,由于网络异常或者gc导致拉取失败,会自动重试,改参数就是配置重试次数,在数据量达到十亿、百亿级别的时候,最好调大该参数以增加稳定性,默认是3次,建议设置为10到20。
spark.shuffle.io.maxRetries=3
19、该参数是 spark.shuffle.io.maxRetries的重试间隔,默认是0.5s。
spark.shuffle.io.retryWait=500
20、shuffle read拉取数据时的缓存区大小,也就是一次拉取的数据大小,默认64,计算单位是M,要注意的它是从n个节点上,一次总共拉取64M数据,而不是从单个节点获取64M。并且它拉取数据时,并行的发送n个请求,每个请求拉取的最大长度是 64M / n,但是实际拉取时都是以block为最小单位的,所以实际获取的有可能会大于64M / n。所以这个配置就有点迷,属于理论上不行,但实际由于block大小而不得不行的配置
spark.reducer.maxSizeInFlight=64
在你对上面缓存区的大小做修改的时候,不要设置的太大,因为要考虑下面的这个配置
spark.reducer.maxReqsInFlight=Int.MaxValue
该配置用来限制每个批次拉数据时,能够发出的请求数上限,默认是scala中Int类型的最大值,一般不另行改动,但是如果你缓存区大小设置的不合理,或者碰上任务生产的中间文件普遍不大,造成spark为了靠近你设置的缓存区大小文件请求一次性发出去很多,这就会造成大量的网络IO导致任务失败,遇到这种情况,要先使用文件聚合,然后考虑AQE、最后调整任务资源,因为前面两个对资源的消耗是有一定的影响的,总之再次就是想告诉你有这种顾虑存在,至于这个上限限制一般不改
21、spark允许你限制每个reduce任务,能够对执行计划中的某个datanode上获取最多多少个数据块,不过一般遇不到改的情况,和上面缓存区面临的请求数一样是一个要知道的概念
spark.reducer.maxBlocksInFlightPerAddress=Int.MaxValue
22、修改系列化方式,这里的序列化是针对shuffle、广播和rdd cache的序列化方式,默认使用java的序列化方式org.apache.spark.serializer.JavaSerializer性能比较低,所以一般都使用org.apache.spark.serializer.KryoSerializer ,至于spark task的序列化由参数spark.closure.serializer配置,目前只支持JavaSerializer。
spark.serializer=org.apache.spark.serializer.JavaSerializer
23、如果你的数据类型是Parquet,且使用spark计算引擎处理hive数据,要注意这个配置,用来决定是否采用spark自己的Serde来解析Parquet文件;Spark SQL为了更好的性能,在读取hive metastore创建的parquet文件时,会采用自己Parquet Serde,而不是采用hive的Parquet Serde来序列化和反序列化,由于两者底层实现差异比较大,所以很容易造成null值和decimal精度问题,默认为true,设为false即可(会采用与hive相同的Serde)。
spark.sql.hive.convertMetastoreParquet=true
如果你操作spark要对hive表的Parquet类型数据写入的时候一定要注意下面的配置。
spark.sql.parquet.writeLegacyFormat
这个参数用来决定是否使用hive的方式来写Parquet文件,这是由于对数据精度的把控上,两个计算框架不一样,比如decimal精度上两者的实现就有差别,导致hive读取spark创建的Parquet文件会报错,在hive中decimal类型是固定的用int32来表示,而标准的parquet规范约定,根据精度的不同会采用int32和int64来存储,而spark就是采用的标准的parquet格式,所以对于精度不同decimal的,底层的存储类型有变化,所以使用spark存储的parquet文件,在使用hive读取时报错,因此要将
spark.sql.parquet.writeLegacyFormat(默认false)
配置设为true,即采用与hive相同的format类来写parquet文件
24、和上面的Parquet一样,orc数据spark和hive的底层实现也不太一样,因此如果你用spark处理hive的orc数据,要注意下面的配置
spark.sql.hive.convertMetaStoreOrc=true
这个配置用来决定spark读取orc的时候是否转换成spark内部的Parquet格式的表,如果你的orc数据来自于hive,就要设置为false,如果为true发生兼容性问题的概率很大
spark.sql.orc.impl=native
该配置决定spark用那种方式写入orc数据,默认的native,是内置的,但是如果数要流向hive,就要配置成
hive
orc.force.positional.evolution=true
这个配置决定spark是否强制解析orc数据,这里的强制说的是,由于orc是列式存储,在不同版本之间很容易发生字段底层存储的顺序不同,或其他不兼容问题,为true时,意味着强制解析这些数据,为数据分析提供了一定的兼容性保证。
hive.merge.orcfile.stripe.level=true
这个参数用来控制orc数据写入时进行的合并策略,为true使用stripe级别进行,而当该参数设置为false时,合并操作会在文件级别进行。这种合并操作是通过启动一个独立的map-reduce任务来实现的,旨在优化存储和提升查询效率。具体来说,ORC文件的合并有助于减少小文件的数量,从而避免因大量小文件而导致的处理效率低下问题。此外,ORC文件的索引结构(如行组级别、条带级别的索引)使得查询时能够快速定位到需要的数据块,避免了不必要的磁盘和网络I/O,进而提高了查询效率。
这种配置对于处理大量小文件特别有用,尤其是在数据仓库环境中,可以显著提升数据处理的速度和效率。通过合理地配置这个参数,可以根据具体的业务需求和数据特点,优化Hive处理ORC文件的策略,从而达到更好的性能表现。
注意:操作orc和Parquet格式时,一定要带上上面的参数,否则轻则数据精度丢失,重则不可识别数据文件导致任务在读取文件阶段就失败了
25、任务重试次数,这个配置在工作中不同的开发版技术环境是不同的值,如果有需要可以更改,默认的原生栈是2
spark.yarn.maxAppAttempts=2
26、有的时候spark任务,当你不太愿意给再多的资源时候,但任务缺失由于数据太多,比如数据块拉去比较耗时之类的,会触发任务超时,这个时候你可以设置下面的两个参数,把超时时间延长
spark.executor.heartbeatInterval=20s
spark.network.timeout=600s
27、推测执行,这个配置和hive任务的推测执行一样的目的,为了任务的稳步运行,一定要确保是关闭的
spark.speculation=false
版权归原作者 尘世壹俗人 所有, 如有侵权,请联系我们删除。