eBay 基於 Apache Kyuubi 構建統一 Serverless Spark 閘道器的實踐

語言: CN / TW / HK

本文來自 eBay 軟體工程師、Apache Kyuubi PPMC Member王斐在Apache SeaTunnel & Kyuubi 聯合 Meetup的分享,介紹了Apache Kyuubi(Incubating)的基本架構和使用場景,eBay基於自身的需求對Kyuubi所做的增強,以及如何基於Kyuubi構建Unified & Serverless Spark Gateway。

Kyuubi是什麼

首先介紹一下Kyuubi。Kyuubi是一個分散式的Thrift JDBC/ODBC server,支援多租戶和分散式等特性,可以滿足企業內諸如ETL、BI報表等大資料場景的應用,目前的主要方向是依託本身的架構設計,圍繞各類主流計算引擎,打造一個Serverless SQL on Lakehouse服務,目前支援的引擎有Spark、Flink、Trino(也就是Presto)。我今天的主題是圍繞Kyuubi和Spark, 關於其它計算引擎這裡不再展開。

對於Spark,Kyuubi有HiveServer2的API,支援Spark多租戶,然後以Serverless的方式執行。HiveServer2是一個經典的JDBC服務,Spark社群也有一個類似的服務叫做Spark Thrift Server。這裡介紹一下Spark Thrift Server和Kyuubi的對比。

Spark Thrift Server可以理解為一個獨立執行的Spark app,負責接收使用者的SQL請求, SQL的編譯以及執行都會在這個app裡面去執行,當用戶的規模達到一定的級別,可能會有一個單點瓶頸。

對於Kyuubi,我們可以看右邊這張圖,有一個紅色的使用者和一個藍色的使用者,他們分別有一個紅色的Spark app和一個藍色的Spark app,他們的SQL請求進來之後,SQL的編譯和執行都是在對應的app之上進行的,就是說Kyuubi Server只進行一次SQL請求的中轉,把SQL直接傳送給背後的Spark app。

對於Spark Thrift Server來講,它需要儲存結果以及狀態資訊,是有狀態的,所以不能支援HA和LB。而Kyuubi不儲存結果,幾乎是一個無狀態的服務,所以Kyuubi支援HA和LB,我們可以增加Kyuubi Server的個數來滿足企業的需求。所以說Kyuubi是一個更好的Spark SQL Gateway。

Kyuubi的架構分為兩層,一層是Server層,一層是Engine層。Server層和Engine層都有一個服務發現層,Kyuubi Server層的服務發現層用於隨機選擇一個Kyuubi Server,Kyuubi Server對於所有使用者來共享的。Kyuubi Engine層的服務發現層對使用者來說是不可見的,它是用於Kyuubi Server去選擇對應的使用者的Spark Engine,當一條使用者的請求進來之後,它會隨機選擇一個Kyuubi Server,Kyuubi Server會去Engine的服務發現層選擇一個Engine,如果Engine不存在,它就會建立一個Spark Engine,這個Engine啟動之後會向Engine的服務發現層去註冊,然後Kyuubi Server和Engine之間的再進行一個Internal的連線,所以說Kyuubi Server是所有使用者共享,Kyuubi Engine是使用者之間資源隔離。

Kyuubi支援一些Engine的共享級別,它是基於隔離和資源之間的平衡。在eBay我們主要使用到了USER 和CONNECTION級別。首先對於CONNECTION級別,對於使用者的每次連線都會創造一個新的app,也就是一個Kyuubi Engine,適用於ETL場景,ETL的workload比較高,需要一個獨立的app去執行;對於USER級別,我們可以看到這裡有兩個user,一個叫Tom,一個叫Jerry,Tom的兩個client連線Kyuubi Server,會連線到同一個屬於Tom的Kyuubi Engine,USER級別適用於ad-hoc場景,就是對於同一個使用者所有的連線都會到同一個Kyuubi Engine去執行,而對Jerry的所有請求都會到Jerry的Kyuubi Engine去執行。

對USER共享級別Kyuubi做了一些加強,引入了一個Engine POOL的概念,就像程式設計裡面的執行緒池一樣,我們可以建立一個Engine的pool,pool裡面有編號,比如說這裡Tom建立了兩個pool,叫做pool-a和pool-b,編號為pool-a-0,pool-a-1,如果說在客戶端請求的時候直接指定這個pool的名字,Kyuubi server會從這個pool裡面去隨機選擇一臺Engine執行;如果Tom在請求的時候不僅指定pool的名字,還指定了這個Engine在pool裡面的索引,比如說這裡指定pool-b-0,Kyuubi Server會從這個pool-b裡面選擇編號為0的Engine去做計算。對應的引數為kyuubi.engine.share.level.subdomain.

這在eBay裡面為BI工具整合提供了很大的便利,因為eBay,每個分析師團隊可能用同一個賬號去執行資料分析,BI工具會根據使用者的IP去建立一個Kyuubi Engine,因為每個分析師需要的引數配置可能是不一樣的,比如說他們的記憶體的配置是不一樣的,BI工具就可以建立一個這樣的engine pool,然後儲存使用者的IP和所建立Engine 索引的一個mapping,然後在這個使用者的請求過來的時候,根據BI工具儲存的IP對映關係,去找到該使用者所建立的Engine,或者是說一個團隊裡面很多人使用一個pool,可以預建立許多Spark app,讓這一個組裡面的人可以隨機選擇一個Engine去做執行,這樣可以加大併發度。 同時也可以作為USER共享級別下面的一個標籤用於標註該引擎的用途,比如說我們可以給beeline場景和java JDBC應用使用場景建立USER共享級別下的不同engine pool,在不同使用場景下使用不同的engine pool, 互相隔離。

前面提到了不同的Engine共享級別,有的是為每個連線建立一個Spark App,有的是為一個使用者建立一個或者多個Kyuubi Engine,你可能會擔心資源浪費,這裡講一下Kyuubi Engine對資源的動態管理。首先,一個Kyuubi Engine,也就是一個Spark app,分為Spark driver和Spark executor,對於executor,Spark本身就有一個executor dynamic allocation機制,它會根據當前Spark job的負載來決定是否向叢集申請更多的資源,或者是說將目前已申請的資源返還給叢集。所說我們在Kyuubi Server層加一些限制,比如強制開啟這個executor dynamic allocation,並且把在空閒時候最小的executor數量設為0,也就是說當一個app非常空閒的時候只有一個driver帶執行,避免浪費資源。除了executor層的動態回收機制,Kyuubi 為driver層也加了資源回收機制。對於CONNECTION分享級別,Kyuubi Engine只有在當前連線才使用,當連線關閉的時候Spark driver會直接被回收掉。對USER級別的共享,Kyuubi 有一個引數kyuubi.session.engine.idle.timeout 來控制engine的最長空閒時間,比如說我們將空閒時間設定為12小時,如果12個小時之內都沒有請求連線到這個Spark app上,這個Spark app就會自動結束,避免資源浪費。

Kyuubi的使用場景

下面講一下Use Case。目前Kyuubi支援了SQL語言和Scala語言的執行,也可以把SQL和Scala寫在一起去跑。因為SQL是一種非常使用者友好的語言,它可以讓你不用瞭解Spark內部的原理,就可以使用簡單的SQL語句去查詢資料,但是它也有一定的侷限性;而Scala語言需要一定的門檻,但它非常的靈活,我們可以去寫程式碼或者去操縱一些Spark DataFrame API。

舉一個例子,就是可以在一個SQL檔案或者一個notebook裡面去混合程式設計。首先用SQL語句建立了一張訓練的資料表,在建立表之後通過SET語句把語言模式設為Scala,然後開始用Scala去寫程式碼,這裡是用一個kMeans把訓練資料進行處理,處理完之後把輸出儲存在一張表裡面,再把語言模式切換到SQL,繼續用SQL去處理。這樣非常方便,我們可以結合SQL、Scala的優點,基本上可以解決資料分析裡面的大部分的case。我們也在Kyuubi JDBC裡面提供了一個非常友好的介面,可以直接呼叫KyuubiStatement::ExecuteScala去執行Scala語句。

Kyuubi在eBay的實踐

eBay需求背景

我們Hadoop team管理了很多個Hadoop叢集,這些叢集分佈在不同的資料中心,有不同的用途,有一套統一的基於KDC和LDAP的許可權校驗。

剛開始引入Kyuubi的時候,我們就在想要為每個叢集都部署一個Kyuubi服務嗎?如果這樣我們可能要部署三四個Kyuubi服務,在升級的時候需要進行重複操作,管理起來很不方便,所以我們就想是否能夠用一套Kyuubi來服務多個Hadoop叢集。

eBay對Kyuubi的增強

下圖就是我們為這個需求所做的一些增強。首先,因為我們是支援KDC和LDAP認證的,我們就讓Kyuubi同時支援Kerberos和Plain型別的許可權認證,並對Kyuubi Engine的啟動、Kyuubi的Beeline做了些優化,然後我們擴充套件了一些Kyuubi的thrift API支援上傳下載資料。針對前面說的要用一個Kyuubi去訪問多個Hadoop叢集,我們加了一個cluster selector的概念,可以在連線的時候指定引數,讓請求路由到對應的叢集。還有就是我們也在完善RESTfull API,已經為Kyuubi支援了SPNEGO和BASIC 的RESTfull API許可權認證。此外我們也在做RESTfull API去跑SQL Query 和 Batch job的一些工作。圖中打編號的是已經回饋社群的一些PR。

這裡講一下2、3、4,對Kyuubi的Thrift RPC的一些優化。首先因為Thrift RPC本身是針對HiveServer2來設計的,HiveServer2/Spark Thriftserver2裡面建立一個連線是非常快的。而在Kyuubi裡面建立一個連線的話,首先要連線到Kyuubi Server,Kyuubi Server要等到和遠端的Kyuubi Engine建立連線完成之後,才能把結果返回給客戶端。

如果Kyuubi Engine一開始不存在,而且在啟動Kyuubi Engine的時候由於資源問題,或者是有一些引數設定不對,比如說他設定了無效的spark.yarn.queu,導致出現錯誤的話,中間可能會有一分鐘或者說幾十秒的延遲,客戶端要一直等,在等的過程中也沒有任何的log返回給客戶端。我們就針對這個做了一些非同步的OpenSession,將OpenSession分為兩部分,第一步是連線到Kyuubi Server,Kyuubi Server再非同步啟動一個LaunchEngine Operation,之後立即把Kyuubi Server連線給客戶端,這樣客戶端可以做到一秒鐘就可以連線到Kyuubi Server。但是他的下條語句以及進來之後,會一直等到這個Engine初始化好之後才開始執行。其實這也是我們的PM的一個需求,即使第一條語句執行時間長一點也沒關係,但是連線是一定要很快,所以我們就做了這樣一個工作。

因為Hive Thrift RPC是一個廣泛應用而且非常使用者友好的RPC,所以我們在不破壞它的相容性的情況下基於Thrift RPC做了一些擴充套件。首先對於ExecuteStatement這種請求及返回結果,它會在API裡面返回一個OperationHandle,再根據OperationHandle獲取當前Operation的狀態和日誌。因為我們前面已經把OpenSession拆成了OpenSession加上一個LaunchEngine Operation,所以我們想把LaunchEngine Operation的一些資訊,通過OpenSession request這個configuration map把它返回去,我們是把一個OperationHandler分為兩部分,一部分是guid,另一部分是secret,放到OpenSessionResp 的configuration Map裡面。

然後在拿到OpenSessionResp之後就可以根據這個configuration拼出Launch Engine Operation對應的OperationHandler,然後再根據它去拿這個LaunchEngine的日誌和狀態。

下面是一個效果,我們在建立Kyuubi連線的時候可以實時知道spark-submit的過程中到底發生了什麼。比如說使用者將spark.yarn.queue設定錯了,或者說由於資源問題一直在等待,都可以清楚的知道這中間發生了什麼,不需要找平臺維護人員去看日誌,這樣既讓使用者感到極為友好,也減少了平臺維護人員的efforts。

構建Unified & Serverless Spark Gateway

前面說到要用一個Kyuubi服務來服務多個叢集,我們就基於Kyuubi構建了一個Unified & Serverless Spark Gateway。Unified是說我們只有一個endpoint,我們是部署在Kubernetes之上的,使用Kubernetes的LB作為Kyuubi Server的服務發現,endpoint的形式就是一個Kubernetes的LB加上一個埠,比如說kyuubi.k8s-lb.ebay.com:10009,要服務多個叢集,我們只需要在JDBC URL裡面加上一個引數kyuubi.session.cluster,指定cluster name,就可以讓他的請求到指定的叢集去執行。關於許可權校驗我們也是用Unified的,同時支援Kerberos和LDAP許可權校驗。關於functions(功能)也是Unified的,同時支援Spark-SQL、Spark-Scala以及ETL Spark Job的提交。

關於Serverless, Kyuubi Server部署在Kubernetes之上,是Cloud-native的,而且Kyuubi Server支援HA和LB,Kyuubi Engine支援多租戶,所以對於平臺維護人員來說成本非常低。

這是我們大概的一個部署,對於多叢集我們引入了Cluster Selector的概念,為每個叢集都配了一個Kubernetes ConfigMap檔案,在這個檔案裡面有這個叢集所獨有的一些配置,比如這個叢集的ZooKeeper的配置,叢集的環境變數,會在啟動Kyuubi Engine的時候注入到啟動的程序裡面。

每個叢集的super user的配置也是不一樣的,所以我們也支援了對各個叢集進行super user的校驗。目前Kyuubi支援HadoopFSDelegation token和HiveDelegation token的重新整理,可以讓Spark Engine在沒有keytab的情況下去長執行,而不用擔心token過期的問題。我們也讓這個功能支援了多叢集。

使用者一個請求進來的過程是這樣的:首先他要指定一個Cluster Selector,Kyuubi Server(on Kubernetes)根據這個Selector去找到對應的叢集,連線叢集的ZooKeeper,然後查詢在ZooKeeper裡面有沒有對應的Spark app, 如果沒有就提交一個app到YARN上面(Spark on YARN),Engine在啟動之後會向ZooKeeper註冊,Kyuubi Server和Kyuubi Engine通過ZooKeeper找到Host和Port並建立連線。

Kyuubi從一開始就支援Thrift/JDBC/ODBC API, 目前社群也在完善RESTFul API. eBay也在做一些完善RESTFul API的工作,我們給RESTful API加了許可權校驗的支援,同時支援SPNEGO(Kerberos)和BASIC(基於密碼)的許可權校驗。我們計劃給Kyuubi增加更多的RESTful API。目前已有的是關於sessions的API,比如可以關掉session,主要用於來管理Kyuubi的sessions。我們準備加一些針對SQL以及Batch Job的API。關於SQL就是可以直接通RESTful API提交一條SQL Query,然後可以拿到它的結果以及log。關於Batch Job就是可以通過RESTful API提交一個普通的使用JAR來執行的Spark app,然後可以拿到Spark app的ApplicationId,還有spark-submit的log,這樣可以讓使用者更加方便地使用Kyuubi完成各種常用的Spark操作。

eBay的收益

對使用者來說,他們能夠非常方便地使用Spark服務,可以使用Thrift、JDBC、ODBC、RESTful的介面,它也是非常輕量的,不需要去安裝Hadoop/Spark binary,也不需要管理Hadoop和Spark的Conf,只需要用RESTful或者Beeline/JDBC的形式去連線就好。

對我們平臺開發團隊來說,我們有了一箇中心化的Spark服務,可以提供SQL、Scala的服務,也可以提供spark-submit的服務,我們可以方便地管理Spark版本,不需要將Spark安裝包分發給使用者去使用,可以更好地完成灰度升級,讓Spark版本對於使用者透明化,讓整個叢集使用最新的Spark,也節省叢集的資源,節省公司的成本。此外,平臺的維護也是比較方便的,成本很低,因為我們只需要維護一個Kyuubi服務來服務多個Hadoop叢集。

作者:王斐,eBay 軟體工程師,Apache Kyuubi PPMC Member

附影片回放及PPT下載:

Apache Kyuubi 在 eBay 的實踐 -王斐

延伸閱讀:

Apache Kyuubi 在 T3 出行的深度實踐

Who is using Apache Kyuubi (Incubating)?