直播回放: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氛围,欢迎钉钉扫码加入!