Flink Sql Gateway的原理與實踐

語言: CN / TW / HK

1 背景

我們在使用Flink開發實時任務時,都會用到框架本身提供的DataStream API,這使得使用者不能不用Java或者Scala甚至Python來編寫業務邏輯;這種方式雖然靈活且表達性強,但對使用者具有一定的開發門檻,並且隨著版本的不斷更新,DataStream API也有很多老版本不相容的問題。所以Flink SQL就成了廣大開發使用者的最佳選擇,之所以Flink推出SQL API,主要是因為SQL有如下幾個重要特性:

圖:SQL重要特性

1.宣告式API:

使用者只關心做什麼,而不關心怎麼做;

2.自動優化:

遮蔽底層API的複雜性,自動做優化;

3.簡單易懂:

SQL應用於不同行業和領域,學習成本較低;

4.不易變動:

語法遵循SQL標準規範,不易變動;

5.流批統一:

同樣的SQL程式碼,可以用流和批的方式執行。

雖然Flink提供了SQL能力,但還是有必要基於Flink SQL打造屬於自己的平臺,目前搭建SQL平臺的方式有如下幾種:

1.Flink原生API:

使用Flink提供的SQL API,封裝一個通用的pipeline jar,利用flink shell指令碼工具提交sql任務;

2.Apache Zeppelin:

一款開源產品,利用notebook方式管理sql任務,目前已經與Flink整合,且提供了豐富的SDK;

3.Flink Sql Gateway:

Flink官方出品的一個Sql閘道器,用Rest方式執行Flink Sql。

第一種方式缺乏靈活性,且大量提交任務時,有效能瓶頸;而Zeppelin雖然功能強大,但頁面功能有限,如果要基於Zeppelin打造SQL平臺,要麼使用SDK,要麼對Zeppelin做重度的二次開發;所以Flink Sql Gateway比較適合做平臺化建設,因為它是一個獨立的閘道器服務,方便與公司現有系統整合,完全與其它系統解耦,本文也主要闡述Flink Sql Gateway的實踐與探索。

2 Flink Sql Gateway 簡介

2.1 架構

圖:Flink Sql Gateway 架構

如上圖所示,Flink Sql Gateway的架構比較簡單,主要元件是SqlGatewayEndpoint,它是基於Flink的RestServerEndpoint實現的一個Netty服務,通過自定義實現多種handler來完成sql任務的建立和部署,以及管理的能力。SqlGatewayEndpoint內部主要由SessionManager(會話管理) 組成,SessionManager維護了一個session map,而session內部主要是一些上下文配置和環境資訊。

1) SqlGatewayEndpoint:

基於RestServerEndpoint實現的Netty服務,對外提供Rest Api;

2) SessionManager :

會話管理器,管理session建立與刪除;

3) Session:

一個會話,裡面存放著任務所需要的Flink配置和上下文環境資訊,負責任務的執行;

4) Classpath:

Flink Sql Gateway啟動時會載入flink安裝目錄的classpath,所以flink sql gateway 基本上沒有除flink以外的相關依賴。

2.2 執行流程

sql gateway其實只是一個普通的NIO伺服器,每個Handler都會持有SessionManager的引用,因此可以共同訪問同一個SessionManager物件。當請求到達時,Handler會獲取請求中的引數,如SessionId等,去SessionManager中查詢對應的Session,從而執行提交sql、查詢任務狀態等工作。請求流程如下圖所示:

圖:請求流程

建立 session,這是使用sql gateway的第一步,SessionManager會把使用者傳入的任務執行模式、配置、planner引擎方式等引數封裝成Session物件,放入map中,並返回sessionid給使用者;

使用者持有sessionid,發起sql request的請求,gateway根據sessionid找到對應的Session物件,開始部署sql job到yarn / kubernetes;

2.3 功能

2.3.1  任務部署

Flink Sql Gateway作為Flink的客戶端,任務部署直接運用了Flink的能力,而 Flink目前支援三種部署模式:

1.in Application Mode,

2.in a Per-Job Mode,

3.in Session Mode。

三種模式有如下兩個區別:

1.叢集生命週期和資源隔離:

per-job mode的叢集生命週期與job相同,但有較強的資源隔離保證。

2.應用程式的main()方法是在客戶端還是在叢集上執行:

session mode和per-job mode在客戶端上執行,而application mode在叢集上執行。

圖:三種模式

從以上可以看出,Application Mode為每個應用程式建立一個會話叢集,並在叢集上執行應用程式的 main() 方法,所以它是session mode和per-job的一個折中方案。

目前為止,Flink只支援jar包任務的application mode,所以想要實現sql任務的application mode,需要自己改造實現,後面會講實現方法。

2.3.2 SQL 能力

Flink Sql Gateway支援的Sql語法如下:

圖:Flink Sql Gateway 支援的 Sql 語法

Flink Sql Gateway支援所有Flink Sql語法,但本身也有一些限制:

1.不支援多條sql執行,多條insert into執行會產生多個任務;

2.不完整的set支援,對於set語法支援存在bug;

3.Sql Hit支援不是很友好,寫在sql裡比較容易出錯。

3 平臺化改造

3.1 SQL 的 application mode 實現

前面說到,flink不支援sql任務的application mode部署,只支援jar包任務。jar 包任務的application mode實現如下圖所示:

圖:jar 包任務的 application mode 實現

1.flink-clients解析出使用者的配置和jar包資訊;

2.ApplicationConfiguration裡指定了main方法的入口類名和入參;

3.ApplicationDeployer負責把Jobmanager啟動,並且啟動時執行Flink Application的main方法。

通過以上流程可以看出,要實現sql的application mode,實現通用執行sql的pipeline jar是關鍵:

實現一個執行sql的通用pipeline jar包,並且預先傳到yarn或者k8s,如下所示:

在ApplicationConfiguration 指定

pepeline jar的main方法入口和引數:

3.2 多 Yarn 叢集支援

目前Flink只支援單Yarn環境的任務部署,對於擁有多套Yarn環境的場景,需要部署多套Flink環境,每個Flink對應一個Yarn環境配置;雖然這種方式能解決問題,但並不是最優的解決方案。熟悉Flink應該都知道,Flink使用 ClusterClientFactory的SPI來生成與外部資源系統(Yarn/kubernetes)的訪問介質(ClusterDescriptor),通過ClusterDescriptor可以完成與資源系統的互動,比如YarnClusterDescriptor,它持有YarnClient物件,可以完成與Yarn的互動;所以對於多Yarn環境,我們只要保證YarnClusterDescriptor 裡持有的YarnClient物件與Yarn環境一一對應即可,程式碼如下圖所示:

作者簡介

Zheng OPPO高階資料平臺工程師

主要負責基於Flink的實時計算平臺開發, 對Flink有較豐富的研發經驗, 也曾參與過Flink社群的貢獻。

本文版權歸OPPO公司所有,如需轉載請在後臺留言聯絡。