您的位置:首页 > 科技 > IT业 > SparkSql oom原因以及参数调优+数据倾斜解决方案

SparkSql oom原因以及参数调优+数据倾斜解决方案

2024/12/23 5:15:50 来源:https://blog.csdn.net/qq_40771567/article/details/140634636  浏览:    关键词:SparkSql oom原因以及参数调优+数据倾斜解决方案

1、Spark历史版本对比

spark1 vs spark2 vs spark3 

1、spark1引入内存计算的理念,解决中间结果落盘导致的效率低下。在理想状况下性能可达到MR的100倍。虽然提高了一定的计算效率,但也带来了大量的内存管理问题,典型的如内存oom问题频发。

2、spark2引入了Tungsten引擎,关键算子效率上比Spark1提升了10倍。启用“统一内存管理”,不再使用“静态内存管理”,不再使用“静态内存管理”,oom问题大幅下降

3、spark3启用自适应查询(Adaptive Query Execution)

  • 1.动态合并shuffle partitions

    可以简化甚至避免调整shuffle分区的数量。用户可以在开始时设置相对较多的shuffle分区数,AQE会在运行时将相邻的小分区合并为较大的分区。

  • 2.动态调整join策略

    在一定程度上避免由于缺少统计信息或着错误估计大小(当然也可能两种情况同时存在),而导致执行次优计划的情况。这种自适应优化可以在运行时sort merge join转换成broadcast hash join,从而进一步提升性能

  • 3.动态优化倾斜的join(skew joins)

    skew joins可能导致负载的极端不平衡,并严重降低性能。在AQE从shuffle文件统计信息中检测到任何倾斜后,它可以将倾斜的分区分割成更小的分区,并将它们与另一侧的相应分区连接起来。这种优化可以并行化倾斜处理,获得更好的整体性能。

Spark内存管理:https://zhuanlan.zhihu.com/p/642722131?utm_id=0

2、SparkSql oom的因素

sparksql oom的因素是多样的,一个程序,从输入、处理2个方面进行考量

2.1 输入数据因素

1、文件格式&压缩

目前我司新建表默认使用parquet格式

 对于snappy压缩,hivesql参考表的“parquet.compression”属性决定是否压缩,采用什么压缩方法,sparksql遇到parquet表默认snappy压缩。

  • parquet相比于text、csv格式普遍有3-4倍的压缩比
  • snappy相比于非压缩普遍有3-4倍的压缩比
  • parquet+snappy相比于原始的text、csv达到6-8倍的压缩比,甚至更高

2、数据文件的大小

  数仓集群默认的文件大小是256M,大数据是分布式计算,普遍存在倾斜不均的情形,加之程序自身的控制,造成了数据文件的大小不一。

  尽管spark有输入数据的拆分能力,但扔存在拆分不均匀的可能。例如:原始文件的大小为300M,将拆分为256M+44M 2个文件,他们的数据规模相差5.8倍。按照1个task处理一个文件的设定,分得256M文件的task将更容易出现oom

3、数据的型态分布

  如字段中包含大量的null、枚举字段大量的相同值等,parquet+snappy将导致实际压缩比进一步提升。

4、大字段如xml,json

  string字段类型,如其中存储的是xml、json等字段(字符长度达千、万字符数级的称为大字段),parquet+snappy导致实际压缩比进一步提升

2.2 运行环境因素

  环境因素也是oom的重要原因,表现在spark默认参数、yarn集群稳定性

1、spark默认参数

  定义在集群的spark.defaults中,每一个提交到集群的spark作业都默认使用这套参数。影响oom的主要参数:

  • spark.executor.cores=4 --executor 4cpu,可并行运行4 task

  • spark.executor.memory=12g --executor 12g内存,task间共享(不建议调整)

  • spark.executor.memoryOverhead=2g --executor 2g堆外内存,task间共享(不建议调整)

  • spark.sql.files.maxPartitionBytes=268435456(256m) --读取数据文件时,单个partition或task最大256m、文件超过256m即拆分

  • spark.sql.adaptive.coalescePartitions.initialPartitionNum=none --aqe动态计算shuffle阶段partitions的最大值,none时向下取spark.sql.shuffle.partitions

  • spark.sql.shuffle.partitions=400 --shuffle阶段最大400个partition,数据规模很大建议调整

  在spark ui 中可以观察任务的具体参数值

2、yarn集群稳定性

  yarn集群稳定性包括yarn资源调度、yarn节点负载高低等。

3、Spark sql的参数调优

1、input阶段

  input阶段将表或数据文件加载到executor,先进行解压(如果有,比如snappy)、展开(比如parquet、orc格式),这些都是在内存中进行,内存不足时是不可溢写磁盘的。

  • 数据文件(拆分后大小差异导致oom)

  • 数据文件(拆分后)大小相似,大JSON字段、高压缩比导致oom

 2、shuffle阶段

  当sql出现出现distinct join、groupby等关键字时,spark需要进行shuffle操作。shuffle操作将前一个或多个阶段的分片数据拉取到同一个excutor进行聚合处理(即我们通常说的reduce),如shuffle partition或者说task数过小,oom是大概率时间。

  • shuffle阶段概要:多少个task执行读/写操作

  •  shuffle阶段详情:读/写是否均匀、是否倾斜

 3.1 参数调优

    目前executor默认配置的是12G+2G内存、4核CPU,最大可以有4个task同时并行(每颗CPU执行一个task),这4个task共享14G内存。日常调优可以如下:

  1、spark.executors.cores

    不论在哪个阶段发生oom,降低executor的负载总是可行的。可调整spark.executor.cores=2,即单个executor同时执行2个task,这样每个task大约可使用7G内存。

  2、spark.sql.adaptive.coalescePartitions.initialPartitionNum/spark.sql.shuffle.partitions

    shuffle阶段的oom,说明shuffle partitions数比较小,导致上阶段的“大规模数据”在本阶段“更聚集”。可调整该参数为更大的值,使大规模数据更分散。

  • spark1,没有spark.sql.adaptive.coalescePartitions.initialPartitionNum参数,固定partitions值为400,调大会造成碎片化
  • spark2,没有spark.sql.adaptive.coalescePartitions.initialPartitionNum参数,可调整spark.sql.shuffle.partitions(配合AQE效果不错)
  • spark3,优先调整spark.sql.adaptive.coalescePartitions.initialPartitionNumtions(是AQE的系列参数之一,默认none时向下取spark.sql.shuffle.partitions)

    *注意:更大的shuffle partitions数固然能减轻shuffle阶段的oom概率,但需要向集群申请更多的服务器资源,不可无限调大(建议控制在2000以内)

方案实现原理:增加shuffle read task的数量,可以让原本分配给一个task的多个key分配给多个task,从而让每个task处理比原来更少的数据。举例来说,如果原本有5个 key,每个key对应10条数据,这5个key都是分配给一个task的,那么这个task就要处理50条数据。而增加了shuffle read task以后,每个task就分配到一个key,即每个task就处理10条数据,那么自然每个task的执行时间都会变短了。

3、spark.sql.files.maxPartitionBytes

  对于input阶段出现的oom,通常原因是输入数据的超高压缩比,如:json、xml大字段、null值占比高、枚举字段多等。

  spark.executor.cores值仍不起作用时,可逐步调小spark.sql.files.maxPartitionBytes=134217728(128m)/67108864(64m)

  调小spark.executor.cores、spark.sql.files.maxPartitionBytes值的本质都是减小executor的负载,spark.executor.cores减少了task的并行度,而spark.sql.files.maxPartitionBytes减小了单个task处理的数据规模。所以:

  •   优先调整spark.executor.cores,验证是否扔有oom
  •   确认不起作用后,再调整spark.sql.files.maxPartitionBytes,此时建议恢复spark.executor.cores的值(如不恢复,相当于2*2=4倍的负载降低、对于单executor14g的内存资源是巨大浪费)

  *注意:spark.sql.files.maxPartitionBytes不建议调得过小(比如32m、16m),过小需要向集群申请更多executor及计算资源、集群负载增大,也可能导致数据文件碎片化

4、数据倾斜解决方案

https://download.csdn.net/blog/column/8735546/111667492

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com