前言
Apache Spark 自 2010 年面世，到当初曾经倒退为大数据批计算的首选引擎。而在 2020 年 6 月份公布的 Spark 3.0 版本也是 Spark 有史以来最大的 Release，其中将近一半的 issue 都属于 SparkSQL。这也投合咱们当初的次要场景（90% 是 SQL），同时也是优化痛点和次要性能点。咱们 Erda 的 FDP 平台（Fast Data Platform）也从 Spark 2.4 降级到 Spark 3.0 并做了一系列的相干优化，本文将次要联合 Spark 3.0 版本进行探讨钻研。
为什么 Spark 3.0 可能“神功大成”，在速度和性能方面有质的冲破？本文就为大家介绍 Spark 3.0 中 SQL Engine 的“天榜第一”—— 自适应查问框架 AQE（Adaptive Query Execution）。
AQE，你是谁？
简略来说，自适应查问就是在运行时一直优化执行逻辑。
Spark 3.0 版本之前，Spark 执行 SQL 是先确定 shuffle 分区数或者抉择 Join 策略后，再按布局执行，过程中不够灵便；当初，在执行完局部的查问后，Spark 利用收集到后果的统计信息再对查问布局从新进行优化。这个优化的过程不是一次性的，而是随着查问会一直进行优化, 让整个查问优化变得更加灵便和自适应。这一改变让咱们辞别之前无休止的被动优化。
AQE，你会啥？
理解了 AQE 是什么之后，咱们再看看自适应查问 AQE 的“三板斧”：
- 动静合并 Shuffle 分区
- 动静调整 Join 策略
- 动静优化数据歪斜
动静合并 shuffle 分区
如果你之前应用过 Spark，兴许某些“调优宝典”会通知你调整 shuffle 的 partitions 数量，默认是 200。然而在不同 shuffle 中，数据的大小和散布根本都是不同的，那么简略地用一个配置，让所有的 shuffle 来遵循，显然不是最优的。
分区过小会导致每个 partition 解决的数据较大，可能须要将数据溢写到磁盘，从而减慢查问速度；分区过大又会带来 GC 压力和低效 I/O 等问题。因而，动静合并 shuffle 分区是十分必要的。AQE 能够在运行期间动静调整分区数来达到性能最优。
如下图所示，如果没有 AQE，shuffle 分区数为 5，对应执行的 Task 数为 5，然而其中有三个的数据量很少，任务分配不均衡，节约了资源，升高了解决效率。
而 AQE 会合并三个小分区，最终只执行三个 Task，这样就不会呈现之前 Task 空转的资源节约状况。
动静调整 join 策略
SparkJoin 策略大抵能够分三种，别离是 Broadcast Hash Join、Shuffle Hash Join 和 SortMerge Join。其中 Broadcast 通常是性能最好的，Spark 会在执行前抉择适合的 Join 策略。
例如上面两个表的大小别离为 100 MB 和 30 MB，小表超过 10 MB (spark.sql.autoBroadcastJoinThreshold = 10 MB)，所以在 Spark 2.4 中，执行前就抉择了 SortMerge Join 的策略，然而这个计划并没有思考 Table2 通过条件过滤之后的大小理论只有 8 MB。
AQE 能够基于运行期间的统计信息，将 SortMerge Join 转换为 Broadcast Hash Join。
在上图中，Table2 通过条件过滤后真正参加 Join 的数据只有 8 MB，因而 Broadcast Hash Join 策略更优，Spark 3.0 会及时抉择适宜的 Join 策略来进步查问性能。
动静优化数据歪斜
数据歪斜始终是咱们数据处理中的常见问题。当将雷同 key 的数据拉取到一个 Task 中解决时，如果某个 key 对应的数据量特地大的话，就会产生数据歪斜，如下图一样产生长尾工作导致整个 Stage 耗时减少甚至 OOM。之前的解决办法比方重写 query 或者减少 key 打消数据分布不均，都十分浪费时间且前期难以保护。
AQE 能够查看分区数据是否歪斜，如果分区数据过大，就将其分隔成更小的分区，通过分而治之来晋升总体性能。
没有 AQE 歪斜优化时，当某个 shuffle 分区的数据量显著高于其余分区，会产生长尾 Task，因为整个 Stage 的完结工夫是按它的最初一个 Task 实现工夫计算，下一个 Stage 只能期待，这会明显降低查问性能。
开启 AQE 后，会将 A0 分成三个子分区，并将对应的 B0 复制三份，优化后将有 6 个 Task 运行 Join，且每个 Task 耗时差不多，从而取得总体更好的性能。通过对歪斜数据的自适应重分区，解决了歪斜分区导致的整个工作的性能瓶颈，进步了查询处理效率。
自适应查问 AQE 凭借着本人的“三板斧”，在 1TB TPC-DS 基准中，能够将 q77 的查问速度进步 8 倍，q5 的查问速度进步 2 倍，且对另外 26 个查问的速度进步 1.1 倍以上，这是一般优化无奈设想的傲人战绩！
真的吗？我不信
口说无凭，自适应查问 AQE 的优越性到底是如何实现，咱们“码”上看看。
AQE 参数阐明
#AQE 开关
spark.sql.adaptive.enabled=true #默认 false，为 true 时开启自适应查问，在运行过程中基于统计信息从新优化查问打算
spark.sql.adaptive.forceApply=true #默认 false，自适应查问在没有 shuffle 或子查问时将不实用，设置为 true 将始终应用
spark.sql.adaptive.advisoryPartitionSizeInBytes=64M #默认 64MB, 开启自适应执行后每个分区的大小。合并小分区和宰割歪斜分区都会用到这个参数
#开启合并 shuffle 分区
spark.sql.adaptive.coalescePartitions.enabled=true #当 spark.sql.adaptive.enabled 也开启时，合并相邻的 shuffle 分区，防止产生过多小 task
spark.sql.adaptive.coalescePartitions.initialPartitionNum=200 #合并之前 shuffle 分区数的初始值，默认值是 spark.sql.shuffle.partitions，可设置高一些
spark.sql.adaptive.coalescePartitions.minPartitionNum=20 #合并后的最小 shuffle 分区数。默认值是 Spark 集群的默认并行性
spark.sql.adaptive.maxNumPostShufflePartitions=500 #reduce 分区最大值，默认 500，可依据资源调整
#开启动静调整 Join 策略
spark.sql.adaptive.join.enabled=true #与 spark.sql.adaptive.enabled 都开启的话，开启 AQE 动静调整 Join 策略
#开启优化数据歪斜
spark.sql.adaptive.skewJoin.enabled=true #与 spark.sql.adaptive.enabled 都开启的话，开启 AQE 动静解决 Join 时数据歪斜
spark.sql.adaptive.skewedPartitionMaxSplits=5 #解决一个歪斜 Partition 的 task 个数下限，默认值为 5；spark.sql.adaptive.skewedPartitionRowCountThreshold=1000000 #歪斜 Partition 的行数上限，即行数低于该值的 Partition 不会被当作歪斜，默认值一千万
spark.sql.adaptive.skewedPartitionSizeThreshold=64M #歪斜 Partition 的大小上限，即大小小于该值的 Partition 不会被当做歪斜，默认值 64M
spark.sql.adaptive.skewedPartitionFactor=5 #歪斜因子，默认为 5。判断是否为歪斜的 Partition。如果一个分区 (DataSize>64M*5) || (DataNum>(1000w*5)), 则视为歪斜分区。spark.shuffle.statistics.verbose=true #默认 false，关上后 MapStatus 会采集每个 partition 条数信息，用于歪斜解决
AQE 性能演示
Spark 3.0 默认未开启 AQE 个性，样例 sql 执行耗时 41 s。
存在 Task 空转状况，shuffle 分区数始终为默认的 200。
开启 AQE 相干配置项，再次执行样例 sql。
样例 sql 执行耗时 18 s，快了一倍以上。
并且每个 Stage 的分区数动静调整，而不是固定的 200。无 task 空转状况，在 DAG 图中也能察看到个性开启。
总结
Spark 3.0 在速度和性能方面得晋升引人注目，它的新个性远不止自适应查问一个，当然也不意味着所有的场景都能有显著的性能晋升，还须要咱们联合业务和数据进行摸索和应用。
注：文中局部图片源自于网络，侵删。