Apache Spark 3.1.1 版本發(fā)布，眾多新特性介紹
Apache Spark 3.1.1 版本于美國當(dāng)?shù)貢r(shí)間2021年3月2日正式發(fā)布，這個(gè)版本繼續(xù)保持使得 Spark 更快，更容易和更智能的目標(biāo)，Spark 3.1 的主要目標(biāo)如下：
?提升了 Python 的可用性；?加強(qiáng)了 ANSI SQL 兼容性；?加強(qiáng)了查詢優(yōu)化；?Shuffle hash join 性能提升；?History Server 支持 structured streaming
注意，由于技術(shù)上的原因，Apache Spark 沒有發(fā)布 3.1.0 版本，雖然你可以在 Maven 倉庫看到 Apache Spark 3.1.0 版本，但千萬別使用。Apache Spark 3.1 版本線的第一個(gè)版本是 Apache Spark 3.1.1，它不是一個(gè)穩(wěn)定版。
本文將介紹 Apache Spark 3.1.1 版本的比較重要的特性及改進(jìn)，限于篇幅僅僅是簡單介紹，更加深入的原理可以關(guān)注數(shù)磚近期發(fā)布的博客。另外更加完整的 ISSUE 可以參見 Apache Spark 3.1.1 Release Notes[1] 。
如果想及時(shí)了解Spark、Hadoop或者HBase相關(guān)的文章，歡迎關(guān)注微信公眾號(hào)：iteblog_hadoop
Project Zen
在這個(gè)版本中，Zen 項(xiàng)目的啟動(dòng)是為了從以下三個(gè)方面提高 PySpark 的可用性：
?更加 Python 化（Being Pythonic）；?PySpark 中更好更容易的可用性；?與其他 Python 庫更好的互操作性
作為這個(gè)項(xiàng)目的一部分，這個(gè)版本包括 PySpark 的許多改進(jìn)——從利用 Python 類型提示到重新設(shè)計(jì)的 PySpark 文檔，比較重要的改進(jìn)如下。
支持 Python 類型（Python typing support ）
PySpark 中的 Python 類型支持最初是作為第三方庫 pyspark-stubs 來創(chuàng)建的，現(xiàn)在已經(jīng)成為一個(gè)成熟和穩(wěn)定的庫。在這個(gè)版本中，PySpark 正式包含了帶有許多功能（參見SPARK-32681）的 Python 類型提示（Python type hints）。Python 類型提示在 IDE 和 notebooks 中最有用，它使用戶和開發(fā)人員能夠利用無縫的自動(dòng)完成功能，包括最近在 Databricks notebooks 中添加的自動(dòng)完成支持。此外，IDE 開發(fā)人員可以通過 Python 類型提示中的靜態(tài)類型和錯(cuò)誤檢測來提高生產(chǎn)力。
支持依賴關(guān)系管理
PySpark 中的依賴管理支持已經(jīng)完成，對(duì)應(yīng)的文檔也添加完成，以幫助 PySpark 用戶和開發(fā)人員，參見SPARK-33824。以前，PySpark 對(duì)依賴管理的支持是不完整的，它只在 YARN 中有效，并且沒有相關(guān)的操作文檔。在這個(gè)版本中，通過利用 -archive 選項(xiàng)（參見SPARK-33530, SPARK-33615）， Conda、 virtualenv 和 PEX 等包管理系統(tǒng)可以在任何類型的集群中工作。關(guān)于這個(gè)可以看下數(shù)磚的這篇文章 How to Manage Python Dependencies in PySpark 以及對(duì)應(yīng)的文檔。
為 PyPI 用戶提供新的安裝選項(xiàng)
這個(gè)版本為 PyPI 用戶提供了新的安裝選項(xiàng)（參見SPARK-32017）。pip 是安裝 PySpark 最常用的方法之一。然而，上一個(gè)版本只允許在 PyPI 中使用 Hadoop 2，但允許使用 Apache Spark 的其他發(fā)行渠道中的其他選項(xiàng)，如Hadoop 2 和 3。在這個(gè)版本中，作為 Project Zen 的一部分，PyPI 用戶也可以使用所有選項(xiàng)。這使他們能夠從 PyPI 安裝并在現(xiàn)有的任何類型的 Spark 集群中運(yùn)行他們的應(yīng)用程序。
PySpark 相關(guān)的文檔完善
這個(gè)版本 (SPARK-31851) 中引入了 PySpark 的新文檔。之前 PySpark 的文檔很難導(dǎo)航，只包含 API 引用。在這個(gè)版本中，文檔被完全重新設(shè)計(jì)，具有細(xì)粒度的分類和易于導(dǎo)航的層次結(jié)構(gòu)（參見 SPARK-32188）。docstrings 具有 numpydoc 風(fēng)格的更好的人類可讀的文本格式(SPARK-32085)，并且有許多有用的頁面，如如何調(diào)試(SPARK-32186)、如何貢獻(xiàn)和測試(SPARK-32190、SPARK-31851)和使用 live notebook 的快速入門(SPARK-32182)。
ANSI SQL 兼容性
這個(gè)版本為 ANSI SQL 的兼容性增加了額外的改進(jìn)，這有助于簡化從傳統(tǒng)數(shù)據(jù)倉庫系統(tǒng)到 Spark 的工作負(fù)載的遷移，主要如下。
自 Spark 3.0 發(fā)布以來，ANSI 方言模式已經(jīng)被引入并得到了增強(qiáng)。在 ANSI 模式中，如果不是嚴(yán)格地準(zhǔn)守 ANSI SQL 行為，那么 Spark 會(huì)把它弄成與 ANSI SQL 風(fēng)格是一致的。在這個(gè)版本中，當(dāng)輸入無效時(shí)（SPARK-33275），更多的操作符/函數(shù)拋出運(yùn)行時(shí)錯(cuò)誤，而不是返回 NULL。對(duì)顯式類型轉(zhuǎn)換進(jìn)行更嚴(yán)格的檢查也是這次發(fā)布的一部分。當(dāng)查詢包含非法的類型轉(zhuǎn)換時(shí)（例如，日期/時(shí)間戳類型被轉(zhuǎn)換為數(shù)字類型），就會(huì)拋出編譯時(shí)錯(cuò)誤，通知用戶這是無效的轉(zhuǎn)換。ANSI 方言模式仍然處于活躍的開發(fā)中，因此它在默認(rèn)情況下是禁用的，但可以通過設(shè)置 spark.sql.ansi=true 來啟用，我們希望它在即將發(fā)布的版本中保持穩(wěn)定。
這個(gè)版本中添加了各種新的 SQL 特性。添加了廣泛使用的標(biāo)準(zhǔn) CHAR/VARCHAR 數(shù)據(jù)類型，這個(gè)數(shù)據(jù)類型是作為變體 String 類型。增加了更多的內(nèi)置函數(shù)（例如 width_bucket （SPARK-21117）和 regexp_extract_all（SPARK-24884]）。目前內(nèi)置操作符/函數(shù)的數(shù)量已經(jīng)達(dá)到350個(gè)。更多的DDL/DML/utility 命令得到了增強(qiáng)，包括 INSERT（SPARK-32976）、MERGE （SPARK-32030）和EXPLAIN （SPARK-32337）。從這個(gè)版本開始，在Spark WebUI 中，SQL 計(jì)劃將以一種更簡單、更結(jié)構(gòu)化的格式呈現(xiàn)，比如使用 EXPLAIN FORMATTED 展示。
統(tǒng)一 CREATE TABLE SQL 語法已經(jīng)在這次發(fā)布中完成。目前 Spark 維護(hù)兩組 CREATE TABLE 語法。當(dāng)語句中不包含 USING 和 STORED AS 子句時(shí)，Spark使用默認(rèn)的 Hive 文件格式。當(dāng) spark.sql.legacy.createHiveTableByDefault 被設(shè)置為 false (Spark 3.1 版默認(rèn)為 true, Databricks Runtime 8.0 版默認(rèn)為 false)，默認(rèn)的表格式依賴于 spark.sql.sources.default 的設(shè)置 (Spark 3.1 版默認(rèn)為 parquet, Databricks Runtime 8.0版默認(rèn)為 delta)。這意味著在 Databricks Runtime 8.0 中 Delta Lake 現(xiàn)在是默認(rèn)格式，將提供更好的性能和可靠性。下面是一個(gè)演示了當(dāng)用戶沒有顯式指定 USING 或 STORED AS 子句時(shí)使用 CREATE TABLE SQL 例子。
CREATE TABLE table1 (col1 int);CREATE TABLE table2 (col1 int) PARTITIONED BY (partCol int);
下表展示了上面兩個(gè)語句在不同環(huán)境的變化：
Spark 3.0 (DBR 7) or before Spark 3.1 * DBR 8.0 Default Format Hive Text Serde Parquet Delta
注意，我們需要顯示的將 spark.sql.legacy.createHiveTableByDefault 設(shè)置為 false，否則 Apache Spark 將使用 Hive Text Serde。
性能提升
Catalyst 是對(duì)大多數(shù) Spark 應(yīng)用程序進(jìn)行優(yōu)化的查詢編譯器。在 Databricks，每天有數(shù)十億個(gè)查詢被優(yōu)化和執(zhí)行，這個(gè)版本增強(qiáng)了查詢優(yōu)化并加速了查詢處理。
Predicate pushdown
謂詞下推是最有效的性能特性之一，因?yàn)樗梢燥@著減少掃描和處理的數(shù)據(jù)量。Spark 3.1 中完成了各種增強(qiáng):
JSON 和 Avro 數(shù)據(jù)源（參見 SPARK-32346）支持謂詞下推，ORC 數(shù)據(jù)源支持嵌套字段的謂詞下推。Filters 也可以通過 EXPAND 算子進(jìn)行下推 (參見 SPARK-33302)。其他改進(jìn)可以參見 SPARK-32858 和 SPARK-24994。
Shuffle 消除，子表達(dá)式消除和嵌套字段修剪
Shuffle 消除（Shuffle removal），子表達(dá)式消除（subexpression elimination）和嵌套字段修剪（nested field pruning）是另外三個(gè)主要的優(yōu)化特性。Shuffle是最昂貴的操作之一，在某些情況下可以避免 Shuffle （參見 SPARK-31869、SPARK-32282、SPARK-33399），但在消除 Shuffle 后，自適應(yīng)查詢規(guī)劃可能不適用。此外，可以刪除重復(fù)或不必要的表達(dá)式求值（參見 SPARK-33092、SPARK-33337、SPARK-33427、SPARK-33540）以減少計(jì)算量。列修剪可以應(yīng)用于各種操作符（參見 SPARK-29721、SPARK-27217、SPARK-31736、SPARK-32163、SPARK-32059）中的嵌套字段，以減少 I/O 資源的使用，便于后續(xù)的優(yōu)化。
Shuffle-Hash Join (SHJ) 支持所有的 join 類型
從這個(gè)版本開始 Shuffle-Hash Join (SHJ) 支持所有的 join 類型（SPARK-32399），同時(shí)支持相應(yīng)的 codegen execution（SPARK-32421）。與 Shuffle-Sort-Merge Join (SMJ) 不同的是，SHJ 不需要排序，因此當(dāng) join 一個(gè)大表和一個(gè)小表時(shí)，SHJ 的 CPU 和 IO 效率比 SMJ 更高。注意，當(dāng)構(gòu)建端（build side）很大時(shí)，SHJ 可能會(huì)導(dǎo)致 OOM，因?yàn)闃?gòu)建 hashmap 是內(nèi)存密集型的。
Streaming 的改進(jìn)
Spark 是構(gòu)建分布式流處理應(yīng)用程序的最佳平臺(tái)。Databricks 每天有超過10萬億的 records 通過 structured streaming 處理。這個(gè)版本增強(qiáng)了 Structured Streaming 的監(jiān)控、可用性和功能。
為了更好地調(diào)試和監(jiān)控 Structured Streaming 應(yīng)用程序，添加了歷史服務(wù)器（History Server ）支持（參見 SPARK-31953）。在 Live UI 中，添加了更多的 metrics（SPARK-33223）、水印間隔（watermark gap）（參見 SPARK-33224）和更多的狀態(tài)自定義度量（state custom metrics）（參見 SPARK-33287）。
添加了新的 Streaming table APIs，用于讀取和寫 streaming DataFrame 到表中，就像 DataFrameReader 和 DataFrameWriter 中的 table API 一樣。在 Databricks Runtime 中，推薦使用 Delta table 表格式，以實(shí)現(xiàn)精確一次（exactly-once）的語義和更好的性能。
Stream-stream Join 增加了兩種新的 join 類型，這個(gè)版本中包括了 full outer (SPARK-32862) 和 left semi (SPARK-32863) join。在 Apache Spark 3.1 之前，已經(jīng)支持了inner, left outer 以及 right outer stream-stream joins。
其他 Spark 3.1 的改進(jìn)
除了這些新特性，這個(gè)版本還關(guān)注可用性、穩(wěn)定性；改進(jìn)和解決了大約1500個(gè)問題。總共超過200位貢獻(xiàn)者，包括個(gè)人和公司，如 Databricks，谷歌，蘋果，華為，Linkedin，微軟，英特爾，IBM，阿里巴巴，F(xiàn)acebook, Nvidia, Netflix, Adobe 等。在這篇博文中，我們重點(diǎn)介紹了 Spark 中一些關(guān)鍵的 SQL、Python 和 streaming 改進(jìn)，限于篇幅 Apache Spark 3.1 中還有許多其他功能這里并沒有涉及，比如 Spark on Kubernetes GAed, node decommissioning, state schema validation 等，可以到 Apache Spark 3.1.1 Release Notes 和 Spark 的官方文檔查找。
引用鏈接
[1] Apache Spark 3.1.1 Release Notes: https://spark.apache.org/releases/spark-release-3-1-1.html
[2] Introducing Apache Spark? 3.1: https://databricks.com/blog/2021/03/02/introducing-apache-spark-3-1.html
相關(guān)閱讀：