1.背景介紹
Apache Spark(以下簡(jiǎn)稱Spark)是一個(gè)開源的分布式計(jì)算框架,由UC Berkeley AMP Lab開發(fā),可用于批處理、交互式查詢(Spark SQL)、實(shí)時(shí)流處理(Spark Streaming)、機(jī)器學(xué)習(xí)(Spark MLlib)和圖計(jì)算(GraphX)。Spark?使用內(nèi)存加載保存數(shù)據(jù)并進(jìn)行迭代計(jì)算,減少磁盤溢寫,同時(shí)支持 Java、Scala、Python?和 R?等多種高級(jí)編程語言,這使得Spark可以應(yīng)對(duì)各種復(fù)雜的大數(shù)據(jù)應(yīng)用場(chǎng)景,例如金融、電商、社交媒體等。
Spark 經(jīng)過多年發(fā)展,作為基礎(chǔ)的計(jì)算框架,不管是在穩(wěn)定性還是可擴(kuò)展性方面,以及生態(tài)建設(shè)都得到了業(yè)界廣泛認(rèn)可。盡管Apache社區(qū)對(duì)Spark逐步引入了諸如鎢絲計(jì)劃、向量化 Parquet Reader?等一系列優(yōu)化,整體的計(jì)算性能也有兩倍左右的提升,但在 3.0?版本以后,整體計(jì)算性能的提升有所減緩,并且隨著存儲(chǔ)、網(wǎng)絡(luò)以及IO技術(shù)的提升,CPU也逐漸成為Spark計(jì)算性能的瓶頸。如何在Spark現(xiàn)有框架上,增強(qiáng)大數(shù)據(jù)計(jì)算能力,提高CPU利用率,成為近年來業(yè)界的研究方向。
2.開源優(yōu)化方案
Spark本身使用scala語言編寫,整體架構(gòu)基于 JVM?開發(fā),只能利用到一些比較基礎(chǔ)的 CPU?指令集。雖然有JIT的加持,但相比目前市面上的Native向量化計(jì)算引擎而言,性能還是有較大差距。因此考慮如何將具有高性能計(jì)算能力的Native向量引擎引用到 Spark?里來,提升 Spark?的計(jì)算性能,突破 CPU?瓶頸,成為一種可行性較高的解決方案。
隨著Meta在2022年超大型數(shù)據(jù)庫(kù)國(guó)際會(huì)議(VLDB)上發(fā)表論文《Velox:Meta's Unified Execution Engine》,并且Intel創(chuàng)建的Gluten項(xiàng)目基于Apache Arrow數(shù)據(jù)格式和Substrait查詢計(jì)劃的JNI API將Spark JVM和執(zhí)行引擎解耦,從而將Velox集成到Spark中,這使得使用Spark框架+Native向量引擎的大數(shù)據(jù)加速方案成為現(xiàn)實(shí)。
3.DPU計(jì)算卡與軟件開發(fā)平臺(tái)
AI大模型的發(fā)展,金融、電商等領(lǐng)域數(shù)據(jù)處理需求的增加,生活應(yīng)用虛擬化程度的加深,都對(duì)現(xiàn)代化數(shù)據(jù)中心提出嚴(yán)峻的考驗(yàn)。未來數(shù)據(jù)中心的發(fā)展趨勢(shì),逐步演變成CPU + DPU + GPU三足鼎立的情況,CPU用于通用計(jì)算,GPU用于加速計(jì)算,DPU則進(jìn)行數(shù)據(jù)處理。將大數(shù)據(jù)計(jì)算卸載到具有高度定制化和數(shù)據(jù)處理優(yōu)化架構(gòu)的大規(guī)模數(shù)據(jù)計(jì)算DPU卡上,可以有效提高計(jì)算密集型應(yīng)用場(chǎng)景下數(shù)據(jù)中心的性能和效率,降低其成本和能耗。
中科馭數(shù)CONFLUX?-2200D?大數(shù)據(jù)計(jì)算DPU卡主要應(yīng)用于大數(shù)據(jù)計(jì)算場(chǎng)景。CONFLUX?-2200D通過計(jì)算DPU卸載加速,存儲(chǔ)DPU卸載加速和網(wǎng)絡(luò)DPU卸載加速實(shí)現(xiàn)大數(shù)據(jù)計(jì)算性能3-6倍提升。CONFLUX?-2200D是基于中科馭數(shù)自主知識(shí)產(chǎn)權(quán)的KPU(Kernel Processing Unit)架構(gòu)、DOE(Data Offloading Engine)硬件數(shù)據(jù)庫(kù)運(yùn)算卸載引擎和LightningDMA中科馭數(shù)自主知識(shí)產(chǎn)權(quán)的基于DMA的直接內(nèi)存寫入技術(shù)提出的領(lǐng)域?qū)S肈PU卡。能夠滿足無侵入適配、自主可控、安全可靠,支持存算一體、存算分離等不同場(chǎng)景。
中科馭數(shù)HADOS是中科馭數(shù)推出的專用計(jì)算敏捷異構(gòu)軟件開發(fā)平臺(tái)。HADOS?數(shù)據(jù)查詢加速庫(kù)通過提供基于列式數(shù)據(jù)的查詢接口,供數(shù)據(jù)查詢應(yīng)用,目前Spark、PostgreSQL已通過插件的形式適配。支持Java、Scala、C和C++語言的函數(shù)調(diào)用,主要包括列數(shù)據(jù)管理、數(shù)據(jù)查詢運(yùn)行時(shí)函數(shù)、任務(wù)調(diào)度引擎、函數(shù)運(yùn)算代價(jià)評(píng)估、內(nèi)存管理、存儲(chǔ)管理、硬件管理、DMA引擎、日志引擎等模塊,目前對(duì)外提供數(shù)據(jù)管理、查詢函數(shù)、硬件管理、文件存儲(chǔ)相關(guān)功能API。
4.Spark框架+Gluten-Velox向量化執(zhí)行引擎+DPU加速卡
4.1方案簡(jiǎn)介
隨著SSD和萬兆網(wǎng)卡普及以及I/O技術(shù)的提升,Spark用戶的數(shù)據(jù)負(fù)載計(jì)算能力逐漸受到CPU性能瓶頸的約束。由于Spark本身基于JVM的Task計(jì)算模型的CPU指令優(yōu)化,要遠(yuǎn)遠(yuǎn)遜色于其他的Native語言(C++等),再加上開源社區(qū)的Native引擎已經(jīng)發(fā)展得比較成熟,具備優(yōu)秀的量化執(zhí)行能力,這就使得那些現(xiàn)有的Spark用戶,如果想要獲得這些高性能計(jì)算能力就需要付出大量的遷移和運(yùn)維成本。
Gluten解決了這一關(guān)鍵性問題,讓Spark用戶無需遷移,就能享受這些成熟的Native引擎帶來的性能優(yōu)勢(shì)。Gluten最核心的能力就是通過Spark Plugin的機(jī)制,把Spark查詢計(jì)劃攔截并下發(fā)給Native引擎來執(zhí)行,跳過原生Spark不高效的執(zhí)行路徑。整體的執(zhí)行框架仍沿用Spark既有實(shí)現(xiàn),并且對(duì)于Native引擎無法承接的算子,Gluten安排Fallback回正常的Spark執(zhí)行路徑進(jìn)行計(jì)算,從而保證Spark任務(wù)執(zhí)行的穩(wěn)定性。同時(shí)Gluten還實(shí)現(xiàn)了Fallback、本地內(nèi)存管理等功能,使得Spark可以更好利用Native引擎帶來的高性能計(jì)算能力。
Velox是一個(gè)集合了現(xiàn)有各種計(jì)算引擎優(yōu)化的新穎的C++數(shù)據(jù)加速庫(kù),其重新設(shè)計(jì)了數(shù)據(jù)模型以支持復(fù)雜數(shù)據(jù)類型的高效計(jì)算,并且提供可重用、可擴(kuò)展、高性能且與上層軟件無關(guān)的數(shù)據(jù)處理組件,用于構(gòu)建執(zhí)行引擎和增強(qiáng)數(shù)據(jù)管理系統(tǒng)。
由于Velox只接收完全優(yōu)化的查詢計(jì)劃作為輸入,不提供 SQL?解析器、dataframe層、其他 DSL?或全局查詢優(yōu)化器,專注于成為大數(shù)據(jù)計(jì)算的執(zhí)行引擎。這就使得Gluten+Velox架構(gòu)可以各司其職,從而實(shí)現(xiàn)數(shù)據(jù)庫(kù)組件模塊化。
要將Gluten+Velox優(yōu)化過的Spark計(jì)算任務(wù)卸載到DPU卡,還缺少一個(gè)異構(gòu)中間層,為此中科馭數(shù)研發(fā)了HADOS異構(gòu)執(zhí)行庫(kù),該庫(kù)提供列數(shù)據(jù)管理、數(shù)據(jù)查詢運(yùn)行時(shí)函數(shù)、任務(wù)調(diào)度引擎、函數(shù)運(yùn)算代價(jià)評(píng)估、內(nèi)存管理等多種DPU能力的API接口,并且支持Java,C++等多種大數(shù)據(jù)框架語言的調(diào)用,擁有極強(qiáng)的拓展性,以及與現(xiàn)有生態(tài)的適配性。HADOS敏捷異構(gòu)軟件平臺(tái)可以適應(yīng)復(fù)雜的大數(shù)據(jù)軟件生態(tài),在付出較小成本的情況下為多種計(jì)算場(chǎng)景提供DPU算力加速。Spark框架集成Gluten+Velox向量化執(zhí)行引擎,然后使用HADOS平臺(tái),就可以將經(jīng)過向量化優(yōu)化的計(jì)算任務(wù),利用DPU執(zhí)行,從而徹底釋放CPU,實(shí)現(xiàn)DPU高性能計(jì)算。
4.2 DPU算力卸載
velox是由C++實(shí)現(xiàn)的向量化計(jì)算引擎,其核心執(zhí)行框架涵蓋了任務(wù)(Task)、驅(qū)動(dòng)(Driver)和操作器(Operator)等組件。velox將Plan轉(zhuǎn)換為由PlanNode組成的一棵樹,然后將PlanNode轉(zhuǎn)換為Operator。Operator作為基礎(chǔ)的算子,是實(shí)際算法執(zhí)行的邏輯框架,也是實(shí)現(xiàn)DPU計(jì)算卸載的關(guān)鍵。
4.2.1?邏輯框架
Operator作為實(shí)際算法的邏輯框架,承載著各種表達(dá)式的抽象,每一個(gè)Operator中包含一個(gè)或多個(gè)表達(dá)式來實(shí)現(xiàn)一個(gè)復(fù)雜完整的計(jì)算邏輯塊,表達(dá)式的底層是由function來具體實(shí)現(xiàn)。Velox向開發(fā)人員提供了API可以實(shí)現(xiàn)自定義scalar function,通過實(shí)現(xiàn)一個(gè)異構(gòu)計(jì)算版本的function,然后將這個(gè)function注冊(cè)到Velox的函數(shù)系統(tǒng)中,就可以將計(jì)算任務(wù)卸載到DPU卡上。任務(wù)執(zhí)行過程如下圖:
中科馭數(shù)的CONFLUX?-2200D?S?大數(shù)據(jù)計(jì)算加速DPU卡可以實(shí)現(xiàn)列式計(jì)算,并且HADOS平臺(tái)支持C++語言,所以可以直接解析Velox的向量化參數(shù)。對(duì)于列式存儲(chǔ)的數(shù)據(jù),經(jīng)過對(duì)數(shù)據(jù)類型的簡(jiǎn)單處理之后,可以直接交給DPU執(zhí)行計(jì)算任務(wù),免去了數(shù)據(jù)行列轉(zhuǎn)換的性能損失,同時(shí)也降低了DPU計(jì)算資源集成的運(yùn)維難度,大大提高了Velox異構(gòu)開發(fā)的效率。
4.2.2?算子卸載
以我們實(shí)現(xiàn)卸載的Filter算子為例,對(duì)于cast(A as bigint)>1這一具體的表達(dá)式,來探究如何實(shí)現(xiàn)”>”這一二元運(yùn)算符的卸載。
Filter算子的Operator中會(huì)使用有一個(gè) std::unique_ptr<ExprSet> exprs_的變量,用來執(zhí)行過濾和投影的計(jì)算。ExprSet是Filter算子計(jì)算的核心,其本質(zhì)是一顆表達(dá)式樹。cast(A as bigint)>1的表達(dá)式樹以及表達(dá)式樹的靜態(tài)節(jié)點(diǎn)類型如下:
節(jié)點(diǎn)類型 | 作用 |
FieldAccessTypedExpr | 表示RowVector中的某一列,作為表達(dá)式的葉子節(jié)點(diǎn) |
ConstantTypedExpr | 表示常量值,作為表達(dá)式的葉子節(jié)點(diǎn) |
CallTypedExpr |
if/and/or/switch/cast/try/coalesce等 |
CastTypedExpr | 類型轉(zhuǎn)換 |
LambdaTypedExpr | Lambda表達(dá)式,作為葉子節(jié)點(diǎn) |
在表達(dá)式的所有子節(jié)點(diǎn)執(zhí)行完后,會(huì)執(zhí)行applyFunction,說明當(dāng)前表達(dá)式節(jié)點(diǎn)是一個(gè)函數(shù)調(diào)用,然后調(diào)用vectorFunction_的apply來對(duì)結(jié)果進(jìn)行處理,輸入是inputValues_數(shù)組,該數(shù)組長(zhǎng)度與函數(shù)的表達(dá)式葉子節(jié)點(diǎn)數(shù)相等(文中示例表達(dá)式的葉子節(jié)點(diǎn)為2),作為函數(shù)的參數(shù),result為輸出,結(jié)果為VectorPtr,程序流程圖如下:
4.2.3 Fallback
現(xiàn)階段我們只實(shí)現(xiàn)了Filter算子的部分表達(dá)式,后續(xù)還會(huì)繼續(xù)支持更多的算子和表達(dá)式。對(duì)于一些無法執(zhí)行的算子和表達(dá)式,還是需要退回給Velox,交由CPU執(zhí)行,從而保證SQL的正常執(zhí)行。由于處理的是列式數(shù)據(jù),所以回退的執(zhí)行計(jì)劃可以不需要任何處理,就可以直接從HADOS退還給Velox,幾乎無性能損失。
4.2.4 DPU資源管理
HADOS平臺(tái)會(huì)對(duì)服務(wù)器的DPU資源進(jìn)行統(tǒng)一管理。對(duì)于卸載的計(jì)算任務(wù)根據(jù)現(xiàn)有的DPU資源進(jìn)行動(dòng)態(tài)分配,從而實(shí)現(xiàn)計(jì)算資源的高效利用。同時(shí)HADOS平臺(tái)還會(huì)對(duì)計(jì)算任務(wù)中所需的內(nèi)存進(jìn)行合理的分配,動(dòng)態(tài)申請(qǐng)和釋放系統(tǒng)內(nèi)存,從而減少額外的內(nèi)存開銷。
4.3?加速效果
單機(jī)單線程local模式,使用1G數(shù)據(jù)集,僅卸載Filter算子的部分表達(dá)式的場(chǎng)景下,TPC-DS語句中有5條SQL語句,可以將使用開源方案的加速效果提升15-20%左右。q70語句,在開源方案提升100%的基礎(chǔ)上,提升了15%;q89語句,在開源方案提升50%的基礎(chǔ)上,提升了27%;q06在開源方案提升170%的基礎(chǔ)上,提升了13%。
單一運(yùn)算符場(chǎng)景下(SELECT a FROM t WHERE a = 100),使用DPU運(yùn)算符相比 Spark原生的運(yùn)算符的加速比最高達(dá)到12.7。
5.不足和展望
中科馭數(shù)HADOS敏捷異構(gòu)軟件平臺(tái)可以十分輕松地與現(xiàn)有開源大數(shù)據(jù)加速框架相結(jié)合,并且為開源框架提供豐富的算力卸載功能。HADOS平臺(tái)在完美發(fā)揮開源加速框架優(yōu)勢(shì)的前提下,為大數(shù)據(jù)任務(wù)提供硬件加速能力。由于現(xiàn)在我們只實(shí)現(xiàn)了較小部分算子卸載的驗(yàn)證,在執(zhí)行具有復(fù)雜算子操作的SQL時(shí)無法發(fā)揮出DPU的全部實(shí)力,并且因?yàn)殚_源方案在設(shè)計(jì)之處并沒有考慮到使用DPU硬件,所以在磁盤IO,算子優(yōu)化等方面的性能還有待優(yōu)化。后續(xù)我們也會(huì)從一下幾個(gè)方面來進(jìn)一步做特定優(yōu)化:
- 開發(fā)更多較復(fù)雜的算子,例如重量級(jí)的聚合算子會(huì)消耗CPU大量的計(jì)算能力從而影響Spark作業(yè),通過將聚合算子卸載到DPU硬件來解放CPU能力,從而使得加速效果更加明顯;
- 優(yōu)化DPU的磁盤讀寫,讓DPU可以直接讀取硬盤數(shù)據(jù),省去數(shù)據(jù)在服務(wù)器內(nèi)部的傳輸時(shí)間,可以減少數(shù)據(jù)準(zhǔn)備階段的性能損耗;
- RDMA技術(shù),可以直讀取遠(yuǎn)端內(nèi)存數(shù)據(jù),數(shù)據(jù)傳輸內(nèi)容直接卸載到網(wǎng)卡,減少數(shù)據(jù)在系統(tǒng)內(nèi)核中額外的數(shù)據(jù)復(fù)制與移動(dòng),可以減少大數(shù)據(jù)任務(wù)計(jì)算過程中的性能損耗。