大家好，我是Tim。

前一段時(shí)間，快手?jǐn)?shù)據(jù)架構(gòu)團(tuán)隊(duì)開(kāi)源了Blaze項(xiàng)目，它是一個(gè)利用本機(jī)矢量化執(zhí)行來(lái)加速SparkSQL查詢(xún)處理的插件。

用通俗的話(huà)講就是通過(guò)使用Rust重寫(xiě)Spark物理執(zhí)行層來(lái)達(dá)到性能提升的目的。

并且在TPC-DS 1TB的所有測(cè)試中，Blaze相比Spark3.3減少了40%的計(jì)算時(shí)間，降低了近一半的集群資源開(kāi)銷(xiāo)。

此外，在快手內(nèi)部上線的數(shù)倉(cāng)生產(chǎn)作業(yè)也觀測(cè)到了平均30%的算力提升，實(shí)現(xiàn)了較大的降本增效。

今天我們就來(lái)聊一聊這個(gè)Blaze，以及近來(lái)重寫(xiě)Spark執(zhí)行層進(jìn)行提效的諸多項(xiàng)目。

Spark填補(bǔ)了一個(gè)空白

在大數(shù)據(jù)建設(shè)的初期，單臺(tái)機(jī)器的 RAM 是有限且昂貴的，所以在進(jìn)行集群規(guī)模計(jì)算時(shí)唯一可行選擇是基于 MapReduce 的 Hadoop，它可以在諸多廉價(jià)的普通主機(jī)上進(jìn)行集群計(jì)算。

眾所周知，MapReduce 對(duì)磁盤(pán) IO 的負(fù)擔(dān)非常重，而且并沒(méi)有真正發(fā)揮所有 RAM 的價(jià)值。

隨著機(jī)器硬件的發(fā)展，RAM的價(jià)格也大幅降低，這時(shí)Spark提出了彈性分布式數(shù)據(jù)集（RDD），這是一種分布式內(nèi)存抽象，可以讓程序員以容錯(cuò)的方式在大型集群上執(zhí)行內(nèi)存計(jì)算。

Spark 完美地填補(bǔ)了這個(gè)空白。突然間，許多大數(shù)據(jù)處理都可以非常高效地完成。

為什么要重寫(xiě)Spark執(zhí)行層？

而隨著硬件技術(shù)的繼續(xù)發(fā)展，Spark也需要進(jìn)行相應(yīng)的優(yōu)化，來(lái)充分的發(fā)揮出底層硬件提供的能力。

以查詢(xún)計(jì)劃執(zhí)行為例。原有的Spark引擎執(zhí)行一個(gè)查詢(xún)計(jì)劃，往往采用火山模型的方式。

這種上層算子遞歸調(diào)用下層算子獲取并處理元組的方式，存在虛函數(shù)調(diào)用次數(shù)較多、指令或數(shù)據(jù)cache miss率高的缺陷，并且這種一次處理一個(gè)元組的方式無(wú)法使用CPU的SIMD指令進(jìn)行優(yōu)化，從而造成查詢(xún)執(zhí)行效率低下的問(wèn)題。

向量化執(zhí)行就是解決上述問(wèn)題的一種有效手段。

然而，我們都知道Spark是使用Scala寫(xiě)的，和JAVA類(lèi)似是運(yùn)行在JVM上的。

在Java中，與C++或Rust相比，沒(méi)有直接的手動(dòng)向量化特性，實(shí)現(xiàn)向量化都是由JVM自動(dòng)控制的。

例如，JVM會(huì)對(duì)循環(huán)進(jìn)行分析，判斷是否有類(lèi)似于向量化的優(yōu)化機(jī)會(huì)。

如果JVM發(fā)現(xiàn)某個(gè)循環(huán)中其計(jì)算次數(shù)大于一定量級(jí)，且指令可以被SIMD指令集所支持，那么它會(huì)將循環(huán)展開(kāi)為并行操作，從而實(shí)現(xiàn)向量化執(zhí)行。

publicstaticvoidvectorAdd(float[]a,float[]b,float[]result){
for(inti=0;i200;i++){
result[i]=a[i]+b[i];
}
}

如上面的代碼所示，其有可能會(huì)被JVM轉(zhuǎn)換為向量化執(zhí)行。

即使循環(huán)符合向量化的條件，JVM也不能保證一定會(huì)自動(dòng)實(shí)現(xiàn)向量化執(zhí)行。在某些情況下，JVM可能會(huì)選擇跳過(guò)向量化執(zhí)行。

所以，到目前為止，Spark中的性能殺手Shuffle等操作依然采用了行式數(shù)據(jù)處理。

不過(guò)對(duì)于讀寫(xiě)列式文件的算子，如Parquet、Orc等，已經(jīng)實(shí)現(xiàn)了向量化的批量操作。

除此以外，Scala、Java實(shí)現(xiàn)的Spark還會(huì)帶來(lái)垃圾收集（GC）開(kāi)銷(xiāo)，這也是JAVA系語(yǔ)言的通病。

如果垃圾回收的時(shí)間太長(zhǎng)，會(huì)嚴(yán)重影響任務(wù)執(zhí)行的穩(wěn)定性，甚至?xí)徽`識(shí)別為節(jié)點(diǎn)失聯(lián)。

最后，Java還存在較高的內(nèi)存消耗和無(wú)法進(jìn)行低級(jí)別的系統(tǒng)優(yōu)化等問(wèn)題，這都迫使人們一直在嘗試重寫(xiě)Spark執(zhí)行層算子。

一直沒(méi)有開(kāi)源的Photon

其實(shí)對(duì)于重寫(xiě)Spark執(zhí)行層算子，Spark的母公司Databricks早已進(jìn)行了嘗試，并已經(jīng)為其付費(fèi)用戶(hù)提供了向量化的執(zhí)行引擎Photon。

Photon已經(jīng)被應(yīng)用多年，其被定位為適用于 Lakehouse 環(huán)境的矢量化查詢(xún)引擎。在Databricks的內(nèi)部數(shù)據(jù)上，Photon 已將一些客戶(hù)工作負(fù)載加速了 10 倍以上。

業(yè)界也一直有向量化執(zhí)行引擎出現(xiàn)，例如velox、gluten（gluten可以支持velox或ck作為后端）。

velox就是一個(gè)單機(jī)/單節(jié)點(diǎn)的c++的向量化query runtime實(shí)現(xiàn)。

當(dāng)然velox目標(biāo)不只是Spark, 它希望統(tǒng)一替換大數(shù)據(jù)計(jì)算引擎的單節(jié)點(diǎn)runtime，包括Spark、Presto、Pytorch，以取得加速效果。

其對(duì)接Spark就是通過(guò)gluten項(xiàng)目進(jìn)行對(duì)接的，velox目前也已基本在Meta公司內(nèi)部落地。

但其開(kāi)源社區(qū)一直不溫不火，甚至涼涼。

Blaze 借助Arrow DataFusion實(shí)現(xiàn)向量化執(zhí)行引擎

Blaze的實(shí)現(xiàn)原理和Velox、Photon都是大同小異。

即在運(yùn)行時(shí)嘗試使用向量化算子計(jì)算，如果不支持則回退回Spark原來(lái)的計(jì)算算子。

不過(guò)需要注意的是，Blaze是將Spark運(yùn)行時(shí)物理運(yùn)算符轉(zhuǎn)換為 Rust DataFusion的向量化實(shí)現(xiàn)。

關(guān)于DataFusion是什么，可以參考這篇文章：Apache Arrow DataFusion到底是什么?

目前Blaze可能還不支持聚合運(yùn)算符，UDF 或 RDD API，這顯然會(huì)影響 TPC-DS 查詢(xún)的整體運(yùn)行時(shí)間。不過(guò)，在快手內(nèi)部據(jù)傳Blaze中已經(jīng)添加了對(duì)聚合運(yùn)算符的支持。

在開(kāi)源的Blaze已支持Spark3.0和Spark3.3版本，其使用方式和gluten類(lèi)似，通過(guò)在Spark環(huán)境中添加相應(yīng)的Jar包實(shí)現(xiàn)功能擴(kuò)展。

目前從其跑的TPC-DS 查詢(xún)性能測(cè)試上可以看出，Blaze平均提升30%的性能，節(jié)約了40%的集群資源。

然而其問(wèn)題也和Velox一樣。

一方面需要在Spark集群環(huán)境中安裝特定版本的Rust/C++，而且Rust/C++在龐大的集群機(jī)器中可能會(huì)存在各種環(huán)境問(wèn)題。

另一方面，其不支持UDF（當(dāng)然Photon也不支持），在真實(shí)的計(jì)算任務(wù)中可能會(huì)存在各種兼容性問(wèn)題，而導(dǎo)致需要回退到原始的Spark執(zhí)行引擎上，可能會(huì)造成原始任務(wù)的性能倒退。

說(shuō)白了，即不可能全局開(kāi)啟Blaze性能優(yōu)化，目前也只能針對(duì)特定任務(wù)特點(diǎn)用戶(hù)進(jìn)行開(kāi)啟優(yōu)化。

Blaze引擎優(yōu)化定位只是針對(duì)Spark引擎，而且在向量化的實(shí)現(xiàn)上又是基于DataFusion開(kāi)源項(xiàng)目，相比Velox引擎其未來(lái)開(kāi)源的路可能會(huì)比較好走一點(diǎn)，畢竟目標(biāo)沒(méi)那么大嘛。

當(dāng)然Blaze的出現(xiàn)還有一個(gè)作用，也許會(huì)迫使Databricks開(kāi)源他們藏掖已久的Photon，這當(dāng)然是一件好事。

就像當(dāng)年Iceberg迫使Databricks開(kāi)源其收費(fèi)的數(shù)據(jù)湖存儲(chǔ)引擎Delta Lake一樣。

總結(jié)

Spark執(zhí)行算子的向量化是未來(lái)必須要走的路，Blaze項(xiàng)目通過(guò)將Spark物理執(zhí)行層算子轉(zhuǎn)換為Rust Arrow DataFusion向量化算子來(lái)提升性能，目前已在快手內(nèi)部部分業(yè)務(wù)上線，并實(shí)現(xiàn)30%的性能提升。

在快手內(nèi)部的成功，并不一定可以在開(kāi)源社區(qū)獲得成功。

一方面，Spark社區(qū)并不會(huì)允許其并入Spark項(xiàng)目來(lái)獲得更大關(guān)注，另一方面這種優(yōu)化實(shí)現(xiàn)方式在真實(shí)重要的業(yè)務(wù)場(chǎng)景，必然存在很多自定義的函數(shù)或算子，這給其在其他公司數(shù)據(jù)團(tuán)隊(duì)上的落地造成了困難。

這可能需要數(shù)據(jù)團(tuán)隊(duì)具有不破不立的精神，否則其并不能帶來(lái)全平臺(tái)的性能收益，而這顯然會(huì)使得使用Blaze項(xiàng)目所需的成本項(xiàng)異常顯眼。

最后，希望Blaze項(xiàng)目可以成功，至少可以迫使Databricks開(kāi)源其Photon，也希望更多native引擎開(kāi)源來(lái)提升Spark任務(wù)執(zhí)行性能。