查看原文
其他

SparkSQL的自适应执行-Adaptive Execution

点击上方蓝色字体,选择“设为星标

回复”资源“获取更多资源

大数据技术与架构点击右侧关注,大数据开发领域最强公众号!

暴走大数据点击右侧关注,暴走大数据!

Spark SQL 自适应执行优化引擎

背景

Adaptive Execution 将可以根据执行过程中的中间数据优化后续执行,从而提高整体执行效率。核心在于两点

  • 执行计划可动态调整

  • 调整的依据是中间结果的精确统计信息

spark 2.3 开始试验功能

spark 3.0 正式发布 自适应查询执行(Adaptive Query Execution)



如何设置合适的shuffle partition数量?

  • 在Spark SQL中, shufflepartition数可以通过参数spark.sql.shuffle.partition来设置,默认值是200。

  • 如果partition太小,单个任务处理的数据量会越大,在内存有限的情况,就会写文件,降低性能,还会oom

  • 如果partition太大,每个处理任务数据量很小,很快结束,导致spark调度负担变大,中间临时文件多

spark sql 最佳执行计划

  • Spark SQL的Catalyst优化器的核心工作就是选择最佳的执行计划,主要依靠:

    • 早起基于规则的优化器RBO

    • spark2.2 加入基于代价的优化CBO

  • 执行计划在计划阶段确定后,不会改变,如果能够获取运行时信息,就可能得到一个更加的执行计划

数据倾斜如何处理

  • 数据倾斜是指某一个partition的数据量远远大于其它partition的数据,导致个别任务的运行时间远远大于其它任务,因此拖累了整个SQL的运行时间。

  • 常见手段:

    • 增加shuffle partition数量,让热点partition的数据分散一些,但是对于同一个key没有作用

    • 增加 BroadcastHashJoin的阈值,在某些场景下可以把SortMergeJoin转化成BroadcastHashJoin而避免shuffle产生的数据倾斜。

    • 手动过滤倾斜key,加入前缀,join表也对key膨胀处理,再join

spark 能否运行时自动处理join中的数据倾斜


自适应执行架构

基础流程
  • sql -> 解析 -> 逻辑计划 -> 物理计划 -> rdd -> job -> dag -> stage -> task run

    • 一旦执行计划确定,无法更新

自适应划分依据
  • 按照每个reducer处理partition数据内存大小分,每个64m

  • 按照每个reducer处理partition数据条数分,100000条

动态调整执行计划

在运行时动态调整join的策略,在满足条件的情况下,即一张表小于Broadcast阈值,可以将SortMergeJoin转化成BroadcastHashJoin。

  • SortMergeJoin,每个reducer通过网络shuffle读取属于自己的数据;会出现不同程度的数据倾斜问题;

  • BroadcastHashJoin,每一个reducer读取一个mapper的整个shuffle output文件,shuffle读变成了本地读取,没有数据通过网络传输;数据量一般比较均匀,也就避免了倾斜;


动态处理数据倾斜

  • 在运行时很容易地检测出有数据倾斜的partition,当执行某个stage时,我们收集该stage每个mapper 的shuffle数据大小和记录条数

  • 如果某一个partition的数据量或者记录条数超过中位数的N倍,并且大于某个预先配置的阈值,我们就认为这是一个数据倾斜的partition,需要进行特殊的处理


Spark 使用

配置参数

  • org.apache.spark.sql.internal.SQLConf

  • spark.sql.adaptive.enabled=true

    • 倾斜处理开关

  • spark.sql.adaptive.shuffle.targetPostShuffleInputSize

    • 动态调整 reduce 个数的 partition 大小依据。如设置 64MB,则 reduce 阶段每个 task 最少处理 64MB 的数据。默认值为 64MB。

  • spark.sql.adaptive.minNumPostShufflePartitions -- v2.4 有 3.0 已经去掉

    • 动态调整 reduce 个数的 partition 条数依据。如设置 20000000,则 reduce 阶段每个 task 最少处理 20000000 条的数据。默认值为 20000000。

  • spark.sql.adaptive.forceApply -- V3.0

    • 自适应执行在没有需要shuffle或者子查询的时候将不适用,当设为true始终使用

  • spark.sql.adaptive.logLevel --v3.0

    • 自适应执行时产生的日志等级

  • spark.sql.adaptive.advisoryPartitionSizeInBytes -- v3.0

    • 倾斜数据分区拆分,小数据分区合并优化时,建议的分区大小

    • 与spark.sql.adaptive.shuffle.targetPostShuffleInputSize含义相同

  • spark.sql.adaptive.coalescePartitions.enabled -- v3.0

    • 是否开启合并小数据分区默认开启,调优策略之一

  • spark.sql.adaptive.coalescePartitions.minPartitionNum -- v3.0

    • 合并后最小的分区数

  • spark.sql.adaptive.fetchShuffleBlocksInBatch -- v3.0

    • 是否批量拉取blocks,而不是一个个的去取

    • 给同一个map任务一次性批量拉取blocks可以减少io 提高性能

  • spark.sql.adaptive.skewJoin.enabled

    • 自动倾斜处理,处理 sort-merge join中的倾斜数据

  • spark.sql.adaptive.skewJoin.skewedPartitionFactor

    • 判断分区是否是倾斜分区的比例

    • 当一个 partition 的 size 大小大于该值(所有 parititon 大小的中位数)且大于spark.sql.adaptive.skewedPartitionSizeThreshold,或者 parition 的条数大于该值(所有 parititon 条数的中位数)且大于 spark.sql.adaptive.skewedPartitionRowCountThreshold,才会被当做倾斜的 partition 进行相应的处理。默认值为 10

  • spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes

    • 单个分区大于默认256MB

参考

  • https://issues.apache.org/jira/browse/SPARK-23128

  • https://blog.csdn.net/weixin_34006468/article/details/91894261

  • https://www.cnblogs.com/zz-ksw/p/11254294.html



欢迎点赞+收藏+转发朋友圈素质三连


文章不错?点个【在看】吧! 👇

: . Video Mini Program Like ,轻点两下取消赞 Wow ,轻点两下取消在看

您可能也对以下帖子感兴趣

文章有问题?点此查看未经处理的缓存