深入研究 Apache Spark 3.0 的新功能

Posted 2021-04-20 Apache Spark技术交流社区

分享嘉宾 Apache Spark PMC 李潇，就职于 Databricks，Spark 研发部主管，领导 Spark，Koalas，Databricks runtime，OEM 的研发团队，在直播中为大家深入讲解了Apache Spark 3.0的新功能。

直播回放：https://developer.aliyun.com/live/2894

以下是直播内容精华整理。

---

Spark3.0解决了超过3400个JIRAs，历时一年多，是整个社区集体智慧的成果。Spark SQL和Spark Cores是其中的核心模块，其余模块如PySpark等模块均是建立在两者之上。Spark3.0新增了太多的功能，无法一一列举，下图是其中24个相对来说比较重要的新功能，下文将会围绕这些进行简单介绍。

一、Performance

与性能相关的新功能主要有：

• Adaptive Query Execution

• Dynamic Partition Pruning

• Query Complication Speedup

• Join Hints

（一）Adaptive Query Execution

Adaptive Query Execution（AQE）在之前的版本里已经有所实现，但是之前的框架存在一些缺陷，导致使用不是很多，在Spark3.0中Databricks（Spark初创团队创建的大数据与AI智能公司）和Intel的工程师合作，解决了相关的问题。

在Spark1.0中所有的Catalyst Optimizer都是基于规则 (rule) 优化的。为了产生比较好的查询规则，优化器需要理解数据的特性，于是在Spark2.0中引入了基于代价的优化器 （cost-based optimizer），也就是所谓的CBO。然而，CBO也无法解决很多问题，比如：

• 数据统计信息普遍缺失，统计信息的收集代价较高；

• 储存计算分离的架构使得收集到的统计信息可能不再准确；

• Spark部署在某一单一的硬件架构上，cost很难被估计；

• Spark的UDF（User-defined Function）简单易用，种类繁多，但是对于CBO来说是个黑盒子，无法估计其cost。

总而言之，由于种种限制，Spark的优化器无法产生最好的Plan。也正是因为上诉原因，运行期的自适应调整就变得相当重要，对于Spark更是如此，于是有了AQE，其基本方法也非常简单易懂。如下图所示，在执行完部分的查询规划后，Spark可以收集到结果的统计信息，然后利用这些信息再对查询规划重新进行优化。这个优化的过程不是一次性的，而是自适应的，也就是说随着查询规划的执行会不断的进行优化, 而且尽可能地复用了现有优化器的已有优化规则。让整个查询优化变得更加灵活和自适应。

Spark3.0中AQE包括三个主要的运行期自适应功能：

• 可以基于运行期的统计信息，将Sort Merge Join 转换为Broadcast Hash Join；

• 可以基于数据运行中间结果的统计信息，减少reducer数量，避免数据在shuffle期间的过量分区导致性能损失；

• 可以处理数据分布不均导致的skew join。

更多的信息大家可以通过搜索引擎查询了解。

如果你是一个Spark的资深用户，可能你读了很多的调优宝典，其中第一条就是让你的Join变得更快的方法就是尽可能地使用Broadcast Hash Join。比如你可以增加spark.sql.autoBroadcastJoinThreshold 阈值，或者使用 broadcast HINT。但是这基本上属于艺高人胆大。首先，这种方法很难调，一不小心就会Out of Memory，甚至性能变得更差，即使现在产生了一定效果，但是随着负载的变化可能调优会完全失败。

也许你会想：Spark为什么不解决这个问题呢？这里有很多挑战，比如：

• 统计信息的缺失，统计信息的不准确，那么就是默认依据文件大小来预估表的大小，但是文件往往是压缩的，尤其是列存储格式，比如parquet 和 ORC，而Spark是基于行处理，如果数据连续重复，file size可能和真实的行存储的真实大小，差别非常之大。这也是为何提高autoBroadcastJoinThreshold，即使不是太大也可能会导致out of memory；

• Filter复杂、UDFs的使用都会使Spark无法准确估计Join输入数据量的大小。当你的query plan异常大和复杂的时候，这点尤其明显。

其中，Spark3.0中基于运行期的统计信息，将Sort Merge Join 转换为Broadcast Hash Join的过程如下图所示。

也许你还会看到调优宝典告诉你调整shuffle产生的partitions的数量。而当前默认数量是200，但是这个200为什么就不得而知了。然而，这个值设置为多少都不是最优的。其实在不同shuffle，数据的输入大小和分布绝大多数都是不一样。那么简单地用一个配置，让所有的shuffle来遵循，显然是不好的。要设得太小，每个partition的大小就会太大，那么GC的压力就会很大，aggregation和sort会更有可能的去spill数据到磁盘。但是，要是设太大，partition的大小就会太小,partition的数量会大。这个会导致不必要的IO，也让task调度器的压力剧增。那么调度器会导致所有task都变慢。这一系列问题在query plan复杂的时候变得尤为突出，还可能会影响到其他性能，最后耗时耗力却调优失败。

对于这个问题的解决，AQE就有优势了。如下图所示，AQE可以在运行期动态的调整partition来达到性能最优。

此外，数据分布不均是Spark调优的一个疑难杂症，它的表现有多种，比如若干task停滞不前，像是出现了bugs，又比如大量的disk spilling会导致很多节点都无事可做。此外，你也许会看到out of memory这种异常。其解决方法也很多，比如找到skew values然后重写query，或者在join的情况下增加skew keys来消除数据分布不均，但是无论哪种方法，都非常浪费时间，且后期难以维护。AQE解决问题的方式如下，其通过shuffle落地后的中间数据结果判断哪些partition是skew的，如果partition过大，就将其分成若干较小的partition，通过分而治之，总体性能大幅提升。

AQE的发布可以说是一个时代的开始，未来将会更进一步发展，引入更多自适应规则，让Spark可以随着数据分布和特性的变化自动改变Query plan，让更多的query编译静态优化变成运行时的动态优化。

（二）Dynamic Partition Pruning

Dynamic Partition Pruning也是一个运行时的动态优化方法，简单来说就是我们可以通过Query的某些分支的中间结果来避免不必要的partition读取，这种方法是无法通过编译期推测出来的，只能在运行时根据结果来判断，这种方法对数据仓库的star-schema效果非常明显，在TPC-DS获得了非常明显的加速，可以加速2-18倍。

（三）Join Hints

Join Hints是一个非常普遍的数据库的优化策略，在3.0之前已经有了Broadcast hash join，3.0之后的版本加了Sort-merge join、Shuffle hash join和 Shuffle nested loop join，但是要注意谨慎使用，因为数据的特性不同，很难保证一直有效，即使有效，也不代表一直有效，随着时间的变化，你的数据变了，可能会让你的query 变慢，变得不稳定。总体来说上面的四种Join的适用条件和特点如下所示，总而言之，使用Join Hints要谨慎。

二、Richer APIs

Spark3.0简化了开发，不但增加了更多的新功能，也改善了众多现有的功能，让更多的用法成为可能，主要有：

• Accelerator-aware Scheduler

• Built-in Functions

• pandas UDF enhancements

• DELETE/UPDATE/MERGE in Catalyst

（一）pandas UDF enhancements

pandas UDF应该说是PySPark用户中最喜爱的特性之一，对于其功能和性能的提升应该都是喜闻乐见的，其发展历程如下图所示。

最新的pandas UDF和之前的不同之处在于引入了Python Type Hints，现在用户可以使用pandas中的数据类型比如pandas.Series等来表示pandas UDF的种类，不再需要记住原来的UDF类型，只需要指定正确的输入和输出类型即可。此外，pandas UDF可以分为pandas UDF和pandas API。

（二）Accelerator-aware Scheduler

Accelerator-aware Scheduler是加速器的调度支持，狭义上也就是指GPU调度支持。加速器经常用来对特定负载做加速，目前，用户还是需要指定什么应用需要加速器资源，但是在将来我们会支持job或者stage级别的调度。Spark3.0中我们已经支持大多调度器，此外，我们还可以通过Web UI来监控GPU的使用，欢迎大家使用，更多详细资料大家可以到社区学习。

（三）Built-in Functions

为了让Spark3.0更方便实用，Spark社区按照其他的主流，比如数据库厂商等，内嵌了如上图所示的32个常用函数，这样用户就无须自己写UDF，并且速度更快。比如针对map类型，Spark3.0新增加了map_keys和map_values，更加地方便易用。其他新增加的更多内嵌函数大家可以到社区具体了解。

三、Monitoring and Debuggability

Spark3.0也增加了一些对监控和调优的改进，主要有：

• Structured Streaming UI

• DDL/DML Enhancements

• Observable Metrics

• Event Log Rollover

（一）Structured Streaming UI

Structured Streaming是在Spark2.0中发布的，在Spark3.0中加入了UI的配置。新的UI主要包括了两种统计信息：已完成的Streaming查询聚合信息和未完成的Streaming查询的当前信息，包括Input Rate、Process Rate、Batch Duration和Operate Duration。

（二）DDL/DML Enhancements

我们还增加了各种DDL/DML命令，比如EXPLAIN和。
EXPLAIN是性能调优的必备工具，读取EXPLAIN是每个用户的基本功，但是随着系统的运行，EXPLAIN的信息越来越多，而且信息多元、多样，在新的版本中我们引入了新的FORMATTED模式，如下所示，在开头处有一个非常精简的树状图，且之后的每个部分都有很详细的解释，更容易加更多的注意，这就从水平扩展变成了垂直扩展，更加的直观。

（三）Observable Metrics

我们还引入了Observable Metrics用以观测数据的质量。要知道数据质量对于很多Spark应用都是相当重要的，通常定义数据质量的Metrics还是非常容易的，比如用一些聚合参数，但是算出这个Metrics的值就非常麻烦，尤其对于流计算来说。

四、SQL Compatibility

SQL兼容性也是Spark必不可提的话题，良好的兼容性更方便用户迁移到Spark平台，在Spark3.0中新增的主要功能有：

• ANSI Store Assignment

• Overflow Checking

• Reserved Keywords in Parser

• Proleptic Gregorian Calendar

也就是说，这个版本中我们让insert遵守了ANSI Store Assignment，并且增加了运行时的overflow的检查，还提供了一个模式让SQL Parser来准确地遵守ANSI标准的保留字，还切换了Calendar，这样更加符合ANSI的SQL标准。比如说我们想要插入两列数据，类型是int和string，如果将int插入到了string中，还是可以的，不会发生数据精度的损失和数据丢失；但是如果我们尝试将string类型插入到int类型中，就有可能发生数据损失甚至丢失。ANSI Store Assignment+Overflow Checking在输入不合法的时候就会在运行时抛出异常，需要注意的是这个设置默认是关闭的，可以根据个人需要打开。

五、Built-in Data Sources

在这个版本中我们提升了预装的数据源，比如Parquet table，我们可以对Nested Column做Column Pruning和Filter Pushdown，此外还支持了对CSV的Filter Pushdown，还引入了Binary Data Source来处理类似于二进制的图片文件。

六、Extensibility and Ecosystem

Spark3.0继续加强了对生态圈的建设：

• 对Data Source V2 API的持续改善和catalog支持；

• 支持Java 11；

• 支持Hadoop 3；

• 支持Hive 3。

（一）Data Source V2 API+Catalog Support

Spark3.0加上了对Catalog的支持来扩展Data Source API。Catalog plugin API可以让用户注册自己的catalog来实现对元数据的处理，这样可以让Spark用户更简单方便的使用数据源的表。对于没有实现Catalog plugin的数据源，用户需要先注册每个外部数据源的表才能访问，但是实现了Catalog plugin API之后我们只需要注册Catalog，然后就可以直接远程访问和操作catalog的表。对于数据源的开发者来说，什么时候支Data Source V2 API呢？下面是几点建议：

不过这里需要注意，Data Source V2还不是很稳定，开发者可能在未来还需要调整相关API的实现。
大数据的发展相当迅速，Spark3.0为了能更方便的部署，我们升级了对各个组件和环境版本的支持，但是要注意以下事项。

关于生态圈，这里要提一下Koalas，它是一个纯的Python库，用Spark实现了绝大部分的pandas API，让pandas用户除了可以处理小数据，也可以处理大数据。Koalas对于pandas用户来说可以将pandas的代码扩展到大数据处理，使得学习PySpark变得更简单；对于现有的PySpark用户来说，多了更多的选择，可以用pandas API来解决生产力问题。过去一年多，Koalas的下载量是惊人的，在pip的下载量单日已经超过了37000，而且还在不断增长，5月的下载量也达到了85万。Koalas的代码其实不多，主要是API的实现，执行还是由Spark来做，所以Spark性能的提升对于Koalas用户来说是直接受益的。Koalas的发布周期相当频密，目前已经有33个发布，欢迎大家下载使用。

如何读和理解Spark UI对大多数新用户来说是一个很大的挑战，尤其对SQL用户来说，在Spark3.0中我们增加了自己的UI文档https://spark.apache.org/docs/latest/web-ui.html
并且增加了SQL Reference ，https://spark.apache.org/docs/latest/sql-ref.html
等，更详细的文档使得用户上手Spark的时候更加容易，欢迎大家去试一试Spark3.0，感受Spark的强大。

---

关键词：Spark3.0、SQL、PySpark、Koalas、pandas、UDF、AQE

相关内容推荐

---

阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区，定期推送精彩案例，技术专家直播，问答区近万人Spark技术同学在线提问答疑，只为营造纯粹的Spark氛围，欢迎钉钉扫码加入！

对开源大数据和感兴趣的同学可以加小编微信（下图二维码，备注“进群”）进入技术交流微信群。