Spark SQL 是 Spark 生態(tài)系統(tǒng)中處理結(jié)構(gòu)化格式數(shù)據(jù)的模塊。它在內(nèi)部使用 Spark Core API 進(jìn)行處理，但對(duì)用戶的使用進(jìn)行了抽象。這篇文章深入淺出地告訴你 Spark SQL 3.x 的新內(nèi)容。

有了 Spark SQL，用戶可以編寫(xiě) SQL 風(fēng)格的查詢。這對(duì)于精通結(jié)構(gòu)化查詢語(yǔ)言或 SQL 的廣大用戶群體來(lái)說(shuō)，基本上是很有幫助的。用戶也將能夠在結(jié)構(gòu)化數(shù)據(jù)上編寫(xiě)交互式和臨時(shí)性的查詢。Spark SQL 彌補(bǔ)了彈性分布式數(shù)據(jù)集resilient distributed data sets(RDD)和關(guān)系表之間的差距。RDD 是 Spark 的基本數(shù)據(jù)結(jié)構(gòu)。它將數(shù)據(jù)作為分布式對(duì)象存儲(chǔ)在適合并行處理的節(jié)點(diǎn)集群中。RDD 很適合底層處理，但在運(yùn)行時(shí)很難調(diào)試，程序員不能自動(dòng)推斷模式schema。另外，RDD 沒(méi)有內(nèi)置的優(yōu)化功能。Spark SQL 提供了數(shù)據(jù)幀DataFrame和數(shù)據(jù)集來(lái)解決這些問(wèn)題。

Spark SQL 可以使用現(xiàn)有的 Hive 元存儲(chǔ)、SerDes 和 UDF。它可以使用 JDBC/ODBC 連接到現(xiàn)有的 BI 工具。

數(shù)據(jù)源

大數(shù)據(jù)處理通常需要處理不同的文件類型和數(shù)據(jù)源(關(guān)系型和非關(guān)系型)的能力。Spark SQL 支持一個(gè)統(tǒng)一的數(shù)據(jù)幀接口來(lái)處理不同類型的源，如下所示。

文件：

• CSV
• Text
• JSON
• XML

JDBC/ODBC：

• MySQL
• Oracle
• Postgres

帶模式的文件：

• AVRO
• Parquet

Hive 表：

• Spark SQL 也支持讀寫(xiě)存儲(chǔ)在 Apache Hive 中的數(shù)據(jù)。

通過(guò)數(shù)據(jù)幀，用戶可以無(wú)縫地讀取這些多樣化的數(shù)據(jù)源，并對(duì)其進(jìn)行轉(zhuǎn)換/連接。

Spark SQL 3.x 的新內(nèi)容

在以前的版本中(Spark 2.x)，查詢計(jì)劃是基于啟發(fā)式規(guī)則和成本估算的。從解析到邏輯和物理查詢計(jì)劃，最后到優(yōu)化的過(guò)程是連續(xù)的。這些版本對(duì)轉(zhuǎn)換和行動(dòng)的運(yùn)行時(shí)特性幾乎沒(méi)有可見(jiàn)性。因此，由于以下原因，查詢計(jì)劃是次優(yōu)的：

• 缺失和過(guò)時(shí)的統(tǒng)計(jì)數(shù)據(jù)
• 次優(yōu)的啟發(fā)式方法
• 錯(cuò)誤的成本估計(jì)

Spark 3.x 通過(guò)使用運(yùn)行時(shí)數(shù)據(jù)來(lái)迭代改進(jìn)查詢計(jì)劃和優(yōu)化，增強(qiáng)了這個(gè)過(guò)程。前一階段的運(yùn)行時(shí)統(tǒng)計(jì)數(shù)據(jù)被用來(lái)優(yōu)化后續(xù)階段的查詢計(jì)劃。這里有一個(gè)反饋回路，有助于重新規(guī)劃和重新優(yōu)化執(zhí)行計(jì)劃。

Figure 1: Query planning

自適應(yīng)查詢執(zhí)行(AQE)

查詢被改變?yōu)檫壿嬘?jì)劃，最后變成物理計(jì)劃。這里的概念是“重新優(yōu)化”。它利用前一階段的可用數(shù)據(jù)，為后續(xù)階段重新優(yōu)化。正因?yàn)槿绱?，整個(gè)查詢的執(zhí)行要快得多。

AQE 可以通過(guò)設(shè)置 SQL 配置來(lái)啟用，如下所示(Spark 3.0 中默認(rèn)為 false)：

spark.conf.set(“spark.sql.adaptive.enabled”,true)

動(dòng)態(tài)合并“洗牌”分區(qū)

Spark 在“洗牌shuffle”操作后確定最佳的分區(qū)數(shù)量。在 AQE 中，Spark 使用默認(rèn)的分區(qū)數(shù)，即 200 個(gè)。這可以通過(guò)配置來(lái)啟用。

spark.conf.set(“spark.sql.adaptive.coalescePartitions.enabled”,true)

動(dòng)態(tài)切換連接策略

廣播哈希是最好的連接操作。如果其中一個(gè)數(shù)據(jù)集很小，Spark 可以動(dòng)態(tài)地切換到廣播連接，而不是在網(wǎng)絡(luò)上“洗牌”大量的數(shù)據(jù)。

動(dòng)態(tài)優(yōu)化傾斜連接

如果數(shù)據(jù)分布不均勻，數(shù)據(jù)會(huì)出現(xiàn)傾斜，會(huì)有一些大的分區(qū)。這些分區(qū)占用了大量的時(shí)間。Spark 3.x 通過(guò)將大分區(qū)分割成多個(gè)小分區(qū)來(lái)進(jìn)行優(yōu)化。這可以通過(guò)設(shè)置來(lái)啟用：

spark.conf.set(“spark.sql.adaptive.skewJoin.enabled”,true)

Figure 2: Performance improvement in Spark 3.x (Source: Databricks)

其他改進(jìn)措施

此外，Spark SQL 3.x還支持以下內(nèi)容。

動(dòng)態(tài)分區(qū)修剪

3.x 將只讀取基于其中一個(gè)表的值的相關(guān)分區(qū)。這消除了解析大表的需要。

連接提示

如果用戶對(duì)數(shù)據(jù)有了解，這允許用戶指定要使用的連接策略。這增強(qiáng)了查詢的執(zhí)行過(guò)程。

兼容 ANSI SQL

在兼容 Hive 的早期版本的 Spark 中，我們可以在查詢中使用某些關(guān)鍵詞，這樣做是完全可行的。然而，這在 Spark SQL 3 中是不允許的，因?yàn)樗型暾?ANSI SQL 支持。例如，“將字符串轉(zhuǎn)換為整數(shù)”會(huì)在運(yùn)行時(shí)產(chǎn)生異常。它還支持保留關(guān)鍵字。

較新的 Hadoop、Java 和 Scala 版本

從 Spark 3.0 開(kāi)始，支持 Java 11 和 Scala 2.12。 Java 11 具有更好的原生協(xié)調(diào)和垃圾校正，從而帶來(lái)更好的性能。 Scala 2.12 利用了 Java 8 的新特性，優(yōu)于 2.11。

Spark 3.x 提供了這些現(xiàn)成的有用功能，而無(wú)需開(kāi)發(fā)人員操心。這將顯著提高 Spark 的整體性能。