Spark 3.0 的新特性了解吗?

Posted 2021-04-27 深入浅出互联网大厂面经

Core, Spark SQL, Structured Streaming

Highlight

• [Project Hydrogen] Accelerator-aware Scheduler (SPARK-24615)

• Adaptive Query Execution (SPARK-31412)

• Dynamic Partition Pruning (SPARK-11150)

• Redesigned pandas UDF API with type hints (SPARK-28264)

• Structured Streaming UI (SPARK-29543)

• Catalog plugin API (SPARK-31121)

• Java 11 support (SPARK-24417)

• Hadoop 3 support (SPARK-23534)

• Better ANSI SQL compatibility

自适应查询执行（Adaptive Query Execution）

在Spark sql执行物理计划过程中，即使由于缺乏或者不准确的数据统计信息和对成本的错误估算导致生成的初始计划不理想，但是自适应查询执行（Adaptive Query Execution）通过在运行时对查询执行计划进行优化，允许Spark Planner在运行时执行可选的执行计划，这些计划将基于运行时统计数据进行优化，从而提升性能。

由于Spark数据存储和计算是分离的，因此无法预测数据的到达。基于这些原因，对于Spark来说，在运行时自适应显得尤为重要。AQE目前提供了三个主要的自适应优化：

• 动态合并shuffle partitions

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

• 动态调整join策略

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

• 动态优化倾斜的join（skew joins）

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

动态分区裁剪（Dynamic Partition Pruning）

使用"动态分区裁剪"，可以基于运行时推断的信息来进一步进行分区裁剪

正常Spark或Hive在查询时，会根据查询条件与分区字段自动过滤底层的数据文件。但是如果过滤条件没有及时的反映到查询上，就会导致数据被冗余加载。比如左边的是没有动态分区裁剪的情况，两张表进行关联操作，左表包含一个过滤条件，右表需要全表读取。经过动态分区优化后，右表可以直接添加过滤条件，如 id in (select id from lefttable where filter_cond) , 这样可以提前过滤掉右表部分分区的数据

ANSI SQL兼容性（Better ANSI SQL compatibility）

Spark之前的版本的SQL语法和函数跟ANSI标准还是存在一些差异，这次3.0版本更新将缩小和ANSI标准之间的差异，包括增加一些ANSI SQL的函数、区分SQL保留关键字以及内置函数等。

切换到公历（SPARK-26651）
构建Spark自己的日期时间模式定义（SPARK-31408）
介绍用于表插入的ANSI存储分配策略（SPARK-28495）
默认情况下，在表插入中遵循ANSI存储分配规则（SPARK-28885）
添加一个SQLConf spark.sql.ansi.enabled（SPARK-28989）
支持用于聚合表达式的ANSI SQL过滤子句（SPARK-27986）
支持ANSI SQL OVERLAY函数（SPARK-28077）
支持ANSI嵌套方括号注释（SPARK-28880）
整数溢出时引发异常（SPARK-26218）
间隔算术运算的溢出检查（SPARK-30341）

......................

Join hints

尽管社区一直在改进优化器，但仍然不能保证优化器可以在任何场景下做出最优决策——join算法的选择是基于统计和启发式算法。当优化器无法做出最佳选择时，用户可以使用join hints来影响优化器以便让它选择更好的计划

增强的Python API：PySpark和Koalas

Spark 3.0为PySpark API做了多个增强功能：

• 带有类型提示的新pandas API
  pandas UDF最初是在Spark 2.3中引入的，用于扩展PySpark中的用户定义函数，并将pandas API集成到PySpark应用中。但是，随着UDF类型的增多，现有接口就变得难以理解。该版本引入了一个新的pandas UDF接口，利用Python的类型提示来解决pandas UDF类型激增的问题。新接口变得更具Python风格化和自我描述性。
  

• 新的pandas UDF类型和pandas函数API
  该版本增加了两种新的pandas UDF类型，即系列迭代器到系列迭代器和多个系列迭代器到系列迭代器。这对于数据预取和昂贵的初始化操作来说非常有用。此外，该版本还添加了两个新的pandas函数API，map和co-grouped map。更多详细信息请参考：databricks.com/blog/202。
  

• 更好的错误处理
  对于Python用户来说，PySpark的错误处理并不友好。该版本简化了PySpark异常，隐藏了不必要的JVM堆栈跟踪信息，并更具Python风格化。

Hydrogen、流和可扩展性

Spark 3.0完成了Hydrogen项目的关键组件，并引入了新功能来改善流和可扩展性。

• 加速器感知调度
  Hydrogen项目旨在更好地统一基于Spark的深度学习和数据处理。GPU和其他加速器已经被广泛用于加速深度学习工作负载。为了使Spark能够利用目标平台上的硬件加速器，该版本增强了已有的调度程序，使集群管理器可以感知到加速器。用户可以通过配置来指定加速器（详细配置介绍可参考：http://spark.apache.org/docs/latest/）。然后，用户可以调用新的RDD API来利用这些加速器。

• 结构化流的新UI
  结构化流最初是在Spark 2.0中引入的。在Databricks，使用量同比增长4倍后，每天使用结构化流处理的记录超过了5万亿条。

Feature Enhancements

• [Hydrogen] Accelerator-aware Scheduler (SPARK-24615)

• Introduce a complete set of Join Hints (SPARK-27225)

• Add PARTITION BY hint for SQL queries (SPARK-28746)

• Metadata Handling in Thrift Server (SPARK-28426)

• Add higher order functions to scala API (SPARK-27297)

• Support simple all gather in barrier task context (SPARK-30667)

• Hive UDFs supports the UDT type (SPARK-28158)

• Support DELETE/UPDATE/MERGE Operators in Catalyst (SPARK-28351, SPARK-28892, SPARK-28893)

• Implement DataFrame.tail (SPARK-30185)

• New built-in functions

• sinh, cosh, tanh, asinh, acosh, atanh (SPARK-28133)

• any, every, some (SPARK-19851)

• bit_and, bit_or (SPARK-27879)

• bit_count (SPARK-29491)

• bit_xor (SPARK-29545)

• bool_and, bool_or (SPARK-30184)

• count_if (SPARK-27425)

Monitoring and Debuggability Enhancements

• New Structured Streaming UI (SPARK-29543)

• SHS: Allow event logs for running streaming apps to be rolled over (SPARK-28594)

• JDBC tab in SHS (SPARK-29724, SPARK-29726)

• Add an API that allows a user to define and observe arbitrary metrics on batch and streaming queries (SPARK-29345)

• Instrumentation for tracking per-query planning time (SPARK-26129)

• Put the basic shuffle metrics in the SQL exchange operator (SPARK-26139)

• SQL statement is shown in SQL Tab instead of callsite (SPARK-27045)

• Add tooltip to SparkUI (SPARK-29449)

• Improve the concurrent performance of History Server (SPARK-29043)

• EXPLAIN FORMATTED command (SPARK-27395)

• 
  

Other Notable Changes

• Java 11 support (SPARK-24417)

• Hadoop 3 support (SPARK-23534)

• Built-in Hive execution upgrade from 1.2.1 to 2.3.7 (SPARK-23710, SPARK-28723, SPARK-31381)

• Use Apache Hive 2.3 dependency by default (SPARK-30034)

• GA Scala 2.12 and remove 2.11 (SPARK-26132)

• Improve logic for timing out executors in dynamic allocation (SPARK-20286)

• Disk-persisted RDD blocks served by shuffle service, and ignored for Dynamic Allocation (SPARK-27677)

• Acquire new executors to avoid hang because of blacklisting (SPARK-22148)