大數(shù)據(jù)時(shí)代，以O(shè)racle為代表的數(shù)據(jù)庫(kù)中間件已經(jīng)逐漸無(wú)法適應(yīng)企業(yè)數(shù)字化轉(zhuǎn)型的需求，Spark將會(huì)是比較好的大數(shù)據(jù)批處理引擎。而隨著Kubernetes越來(lái)越火，很多數(shù)字化企業(yè)已經(jīng)把在線業(yè)務(wù)搬到了Kubernetes之上，并希望在此之上建設(shè)一套統(tǒng)一的、完整的大數(shù)據(jù)基礎(chǔ)架構(gòu)。那么Spark on Kubernetes面臨哪些挑戰(zhàn)？又該如何解決？

云原生背景介紹與思考

“數(shù)據(jù)湖”正在被越來(lái)越多人提起，盡管定義并不統(tǒng)一，但企業(yè)已紛紛投入實(shí)踐，無(wú)論是在云上自建還是使用云產(chǎn)品。

阿里云大數(shù)據(jù)團(tuán)隊(duì)認(rèn)為：數(shù)據(jù)湖是大數(shù)據(jù)和AI時(shí)代融合存儲(chǔ)和計(jì)算的全新體系。為什么這么說(shuō)？在數(shù)據(jù)量爆發(fā)式增長(zhǎng)的今天，數(shù)字化轉(zhuǎn)型成為IT行業(yè)的熱點(diǎn)，數(shù)據(jù)需要更深度的價(jià)值挖掘，因此確保數(shù)據(jù)中保留的原始信息不丟失，應(yīng)對(duì)未來(lái)不斷變化的需求。當(dāng)前以O(shè)racle為代表的數(shù)據(jù)庫(kù)中間件已經(jīng)逐漸無(wú)法適應(yīng)這樣的需求，于是業(yè)界也不斷地產(chǎn)生新的計(jì)算引擎，以便應(yīng)對(duì)大數(shù)據(jù)時(shí)代的到來(lái)。企業(yè)開(kāi)始紛紛自建開(kāi)源Hadoop數(shù)據(jù)湖架構(gòu)，原始數(shù)據(jù)統(tǒng)一存放在對(duì)象存儲(chǔ)OSS或HDFS系統(tǒng)上，引擎以Hadoop和Spark開(kāi)源生態(tài)為主，存儲(chǔ)和計(jì)算一體。

圖1是基于ECS底座的EMR架構(gòu)，這是一套非常完整的開(kāi)源大數(shù)據(jù)生態(tài)，也是近10年來(lái)每個(gè)數(shù)字化企業(yè)必不可少的開(kāi)源大數(shù)據(jù)解決方案。

圖1 基于ECS的開(kāi)源大數(shù)據(jù)解決方案

主要分為以下幾層：

ECS物理資源層，也就是Iaas層。

數(shù)據(jù)接入層，例如實(shí)時(shí)的Kafka，離線的Sqoop。

存儲(chǔ)層，包括HDFS和OSS，以及EMR自研的緩存加速JindoFS。

計(jì)算引擎層，包括熟知的Spark，Presto、Flink等這些計(jì)算引擎。

數(shù)據(jù)應(yīng)用層，如阿里自研的Dataworks、PAI以及開(kāi)源的Zeppelin，Jupyter。

每一層都有比較多的開(kāi)源組件與之對(duì)應(yīng)，這些層級(jí)組成了最經(jīng)典的大數(shù)據(jù)解決方案，也就是EMR的架構(gòu)。我們對(duì)此有以下思考：

是否能夠做到更好用的彈性，也就是客戶可以完全按照自己業(yè)務(wù)實(shí)際的峰值和低谷進(jìn)行彈性擴(kuò)容和縮容，保證速度足夠快，資源足夠充足。

不考慮現(xiàn)有狀況，看未來(lái)幾年的發(fā)展方向，是否還需要支持所有的計(jì)算引擎和存儲(chǔ)引擎。這個(gè)問(wèn)題也非常實(shí)際，一方面是客戶是否有能力維護(hù)這么多的引擎，另一方面是某些場(chǎng)景下是否用一種通用的引擎即可解決所有問(wèn)題。舉個(gè)例子來(lái)說(shuō)，Hive和Mapreduce，誠(chéng)然現(xiàn)有的一些客戶還在用Hive on Mapreduce，而且規(guī)模也確實(shí)不小，但是未來(lái)Spark會(huì)是一個(gè)很好的替代品。

存儲(chǔ)與計(jì)算分離架構(gòu)，這是公認(rèn)的未來(lái)大方向，存算分離提供了獨(dú)立的擴(kuò)展性，客戶可以做到數(shù)據(jù)入湖，計(jì)算引擎按需擴(kuò)容，這樣的解耦方式會(huì)得到更高的性?xún)r(jià)比。

基于以上這些思考，我們考慮一下云原生的這個(gè)概念，云原生比較有前景的實(shí)現(xiàn)就是Kubernetes，所以有時(shí)候我們一提到云原生，幾乎就等價(jià)于是Kubernetes。隨著Kubernetes的概念越來(lái)越火，客戶也對(duì)該技術(shù)充滿了興趣，很多客戶已經(jīng)把在線的業(yè)務(wù)搬到了Kubernetes之上，并且希望在這種類(lèi)似操作系統(tǒng)上，建設(shè)一套統(tǒng)一的、完整的大數(shù)據(jù)基礎(chǔ)架構(gòu)。所以我們總結(jié)為以下幾個(gè)特征：

希望能夠基于Kubernetes來(lái)包容在線服務(wù)、大數(shù)據(jù)、AI等基礎(chǔ)架構(gòu)，做到運(yùn)維體系統(tǒng)一化。

存算分離架構(gòu)，這個(gè)是大數(shù)據(jù)引擎可以在Kubernetes部署的前提，未來(lái)的趨勢(shì)也都在向這個(gè)方向走。

通過(guò)Kubernetes的天生隔離特性，更好的實(shí)現(xiàn)離線與在線混部，達(dá)到降本增效目的。

Kubernetes生態(tài)提供了非常豐富的工具，我們可以省去很多時(shí)間搞基礎(chǔ)運(yùn)維工作，從而可以專(zhuān)心來(lái)做業(yè)務(wù)。

EMR計(jì)算引擎 on ACK

圖2是EMR計(jì)算引擎 on ACK的架構(gòu)。ACK就是阿里云版本的Kubernetes，在兼容社區(qū)版本的API同時(shí)，做了大量的優(yōu)化。在本文中不會(huì)區(qū)分ACK和Kubernetes這兩個(gè)詞，可以認(rèn)為代表同一個(gè)概念。

圖2 計(jì)算引擎Kubernetes化

基于最開(kāi)始的討論，我們認(rèn)為比較有希望的大數(shù)據(jù)批處理引擎是Spark和Presto，當(dāng)然我們也會(huì)隨著版本迭代逐步的加入一些比較有前景的引擎。

EMR計(jì)算引擎提供以Kubernetes為底座的產(chǎn)品形態(tài)，本質(zhì)上來(lái)說(shuō)是基于CRD+Operator的組合，這也是云原生最基本的哲學(xué)。我們針對(duì)組件進(jìn)行分類(lèi)，分為service組件和批處理組件，比如Hive Metastore就是service組件，Spark就是批處理組件。

圖中綠色部分是各種Operator，技術(shù)層面在開(kāi)源的基礎(chǔ)上進(jìn)行了多方面的改進(jìn)，產(chǎn)品層面針對(duì)ACK底座進(jìn)行了各方面的兼容，能夠保證用戶在集群中很方便的進(jìn)行管控操作。右邊的部分，包括Log、監(jiān)控、數(shù)據(jù)開(kāi)發(fā)、ECM管控等組件，這里主要是集成了阿里云的一些基礎(chǔ)設(shè)施。我們?cè)賮?lái)看下邊的部分：

引入了JindoFS作為OSS緩存加速層，做計(jì)算與存儲(chǔ)分離的架構(gòu)。

打通了現(xiàn)有EMR集群的HDFS，方便客戶利用已有的EMR集群數(shù)據(jù)。

引入Shuffle Service來(lái)做Shuffle 數(shù)據(jù)的解耦，這也是EMR容器化區(qū)別于開(kāi)源方案的比較大的亮點(diǎn)，之后會(huì)重點(diǎn)講到。

這里明確一下，由于本身Presto是無(wú)狀態(tài)的MPP架構(gòu)，在ACK中部署是相對(duì)容易的事情，本文主要討論Spark on ACK的解決方案。

Spark on Kubernetes的挑戰(zhàn)

整體看，Spark on Kubernetes面臨以下問(wèn)題：

我個(gè)人認(rèn)為最重要的，就是Shuffle的流程，按照目前的Shuffle方式，我們是沒(méi)辦法打開(kāi)動(dòng)態(tài)資源特性的。而且還需要掛載云盤(pán)，云盤(pán)面臨著Shuffle數(shù)據(jù)量的問(wèn)題，掛的比較大會(huì)很浪費(fèi)，掛的比較小又支持不了Shuffle Heavy的任務(wù)。

調(diào)度和隊(duì)列管理問(wèn)題，調(diào)度性能的衡量指標(biāo)是，要確保當(dāng)大量作業(yè)同時(shí)啟動(dòng)時(shí)，不應(yīng)該有性能瓶頸。作業(yè)隊(duì)列這一概念對(duì)于大數(shù)據(jù)領(lǐng)域的同學(xué)應(yīng)該非常熟悉，他提供了一種管理資源的視圖，有助于我們?cè)陉?duì)列之間控制資源和共享資源。

讀寫(xiě)數(shù)據(jù)湖相比較HDFS，在大量的Rename，List等場(chǎng)景下性能會(huì)有所下降，同時(shí)OSS帶寬也是一個(gè)不可避免的問(wèn)題。

針對(duì)以上問(wèn)題，我們分別來(lái)看下解決方案。

Spark on Kubernetes的解決方案

Remote Shuffle Service架構(gòu)

Spark Shuffle的問(wèn)題，我們?cè)O(shè)計(jì)了Shuffle 讀寫(xiě)分離架構(gòu)，稱(chēng)為Remote Shuffle Service。首先探討一下為什么Kubernetes不希望掛盤(pán)呢，我們看一下可能的選項(xiàng)：

如果用是Docker的文件系統(tǒng)，問(wèn)題是顯而易見(jiàn)的，因?yàn)樾阅苈徽f(shuō)，容量也是極其有限，對(duì)于Shuffle過(guò)程是十分不友好的。

如果用Hostpath，熟悉Spark的同學(xué)應(yīng)該知道，是不能夠啟動(dòng)動(dòng)態(tài)資源特性的，這個(gè)對(duì)于Spark資源是一個(gè)很大的浪費(fèi)，而且如果考慮到后續(xù)遷移到Serverless K8s，那么從架構(gòu)上本身就是不支持Hostpath的。

如果是Executor掛云盤(pán)的PV，同樣道理，也是不支持動(dòng)態(tài)資源的，而且需要提前知道每個(gè)Executor的Shuffle數(shù)據(jù)量，掛的大比較浪費(fèi)空間，掛的小Shuffle數(shù)據(jù)又不一定能夠容納下。

所以Remote Shuffle架構(gòu)針對(duì)這一痛點(diǎn)、對(duì)現(xiàn)有的Shuffle機(jī)制有比較大的優(yōu)化，圖3中間有非常多的控制流，我們不做具體的討論。主要來(lái)看數(shù)據(jù)流，對(duì)于Executor所有的Mapper和Reducer，也就是圖中的藍(lán)色部分是跑在了K8s容器里，中間的架構(gòu)是Remote Shuffle Service，藍(lán)色部分的Mapper將Shuffle數(shù)據(jù)遠(yuǎn)程寫(xiě)入service里邊，消除了Executor的Task對(duì)于本地磁盤(pán)的依賴(lài)。Shuffle Service會(huì)對(duì)來(lái)自不同Mapper的同一partition的數(shù)據(jù)進(jìn)行merge操作，然后寫(xiě)入到分布式文件系統(tǒng)中。等到Reduce階段，Reducer通過(guò)讀取順序的文件，可以很好地提升性能。這套系統(tǒng)最主要的實(shí)現(xiàn)難點(diǎn)就是控制流的設(shè)計(jì)，還有各方面的容錯(cuò)，數(shù)據(jù)去重，元數(shù)據(jù)管理等等工作。

圖3 Remote Shuffle Service架構(gòu)圖

簡(jiǎn)而言之，我們總結(jié)為以下3點(diǎn)：

Shuffle數(shù)據(jù)通過(guò)網(wǎng)絡(luò)寫(xiě)出，中間數(shù)據(jù)計(jì)算與存儲(chǔ)分離架構(gòu)

DFS 2副本，消除Fetch Failed引起的重算，Shuffle Heavy作業(yè)更加穩(wěn)定

Reduce階段順序讀磁盤(pán)，避免現(xiàn)有版本的隨機(jī)IO，大幅提升性能

Remote Shuffle Service性能

我們?cè)谶@里展示一下關(guān)于性能的成績(jī)，圖4和圖5是Terasort Benchmark。之所以選取Terasrot這種workload來(lái)測(cè)試，是因?yàn)樗挥?個(gè)stage，而且是一個(gè)大Shuffle的任務(wù)，大家可以非常有體感的看出關(guān)于Shuffle性能的變化。

圖4中，藍(lán)色部分是Shuffle Service版本的運(yùn)行時(shí)間，橙色部分是原版Shuffle的運(yùn)行時(shí)間。我們測(cè)試了2T，4T，10T的數(shù)據(jù)，可以觀察到隨著數(shù)據(jù)量越來(lái)越大，Shuffle Service優(yōu)勢(shì)就越來(lái)越明顯。圖5紅框部分說(shuō)明了作業(yè)的性能提升主要體現(xiàn)在Reduce階段，可見(jiàn)10T的Reduce Read從1.6小時(shí)下降到了1小時(shí)。原因前邊已經(jīng)解釋的很清楚了，熟悉Spark shuffle機(jī)制的同學(xué)知道，原版的sort shuffle是M*N次的隨機(jī)IO，在這個(gè)例子中，M是12000，N是8000，而Remote Shuffle就只有N次順序IO，這個(gè)例子中是8000次，所以這是提升性能的根本所在。

圖4 Remote Shuffle Service性能

圖5 Terasort作業(yè)的stage圖

其他方面的重點(diǎn)優(yōu)化

這里講一下EMR在其他方面做的優(yōu)化。

調(diào)度性能優(yōu)化，我們解決了開(kāi)源的Spark Operator的一些不足，對(duì)于Executor pod的很多配置Spark Operator把他做到了Webhook里邊，這個(gè)對(duì)調(diào)度來(lái)說(shuō)是十分不友好的，因?yàn)橄喈?dāng)于在API Server上繞了一圈，實(shí)際測(cè)試下來(lái)性能損耗很大。我們把這些工作直接做到Spark內(nèi)核里邊，這樣避免了這方面的調(diào)度性能瓶頸。然后從調(diào)度器層面上，我們保留了兩種選擇給客戶，包括ACK提供的Scheduler FrameworkV2實(shí)現(xiàn)方式和Yunicorn實(shí)現(xiàn)方式。

讀寫(xiě)OSS性能優(yōu)化，我們這里引入了JindoFS作為緩存，解決帶寬問(wèn)題，同時(shí)EMR為OSS場(chǎng)景提供了Jindo Job Committer，專(zhuān)門(mén)優(yōu)化了job commit的過(guò)程，大大減少了Rename和List等耗時(shí)操作。

針對(duì)Spark本身，EMR積累了多年的技術(shù)優(yōu)勢(shì)，也在TPCDS官方測(cè)試中，取得了很好的成績(jī)，包括Spark性能、穩(wěn)定性，還有Delta lake的優(yōu)化也都有集成進(jìn)來(lái)。

我們提供了一站式的管控，包括Notebook作業(yè)開(kāi)發(fā)，監(jiān)控日志報(bào)警等服務(wù)，還繼承了NameSpace的ResourceQuota等等。

總體來(lái)說(shuō)，EMR版本的Spark on ACK，無(wú)論在架構(gòu)上、性能上、穩(wěn)定性上、易用性方面都有著很大的提升。

Spark云原生后續(xù)展望

從我的視角來(lái)觀察，Spark云原生容器化后續(xù)的方向，一方面是達(dá)到運(yùn)維一體化，另一方面主要希望做到更好的性?xún)r(jià)比。我們總結(jié)為以下幾點(diǎn)：

可以將Kubernetes計(jì)算資源分為固定集群和Serverless集群的混合架構(gòu)，固定集群主要是一些包年包月、資源使用率很高的集群，Serverless是做到按需彈性。

可以通過(guò)調(diào)度算法，靈活的把一些SLA不高的任務(wù)調(diào)度到Spot實(shí)例上，就是支持搶占式ECI容器，這樣可以進(jìn)一步降低成本。

由于提供了Remote Shuffle Service集群，充分讓Spark在架構(gòu)上解耦本地盤(pán)，只要Executor空閑就可以銷(xiāo)毀。配合上OSS提供的存算分離，必定是后續(xù)的主流方向。

對(duì)于調(diào)度能力，這方面需要特別的增強(qiáng)，做到能夠讓客戶感受不到性能瓶頸，短時(shí)間內(nèi)調(diào)度起來(lái)大量的作業(yè)。同時(shí)對(duì)于作業(yè)隊(duì)列管理方面，希望做到更強(qiáng)的資源控制和資源共享。

圖6 Spark on Kubernetes混合架構(gòu)

隨著阿里云2.0時(shí)代的到來(lái)，云原生已經(jīng)逐漸成為新的主角，越來(lái)越多的企業(yè)將會(huì)享受到云原生所帶來(lái)的紅利。數(shù)據(jù)湖計(jì)算引擎云原生容器化，能夠極大地方便客戶上云，提升性?xún)r(jià)比。但是整體上來(lái)講：大數(shù)據(jù)云原生的落地是十分具有挑戰(zhàn)性的，阿里云EMR團(tuán)隊(duì)會(huì)和社區(qū)同伴、合作伙伴一起打造技術(shù)和生態(tài)，共同打造“存算分離，按需擴(kuò)展”、“極致彈性，隨用隨得”、“運(yùn)維閉環(huán)，高性?xún)r(jià)比”的愿景。
編輯：hfy