?摘要：本文整理自網(wǎng)易游戲資深開發(fā)工程師林小鉑在 Flink Forward Asia 2021 平臺(tái)建設(shè)專場的演講。主要內(nèi)容包括：

1. 網(wǎng)易游戲 Flink SQL 發(fā)展歷程
2. 基于模板 jar 的 StreamflySQL v1
3. 基于 SQL Gateway 的 StreamflySQL v2
4. 未來工作?

01網(wǎng)易游戲 Flink SQL 發(fā)展歷程

網(wǎng)易游戲?qū)崟r(shí)計(jì)算平臺(tái)叫做 Streamfly，這個(gè)名字取名自電影《馴龍高手》中的 Stormfly。由于我們已經(jīng)在從 Storm 遷移到 Flink，所以將 Stormfly 中的 Storm 替換成了更為通用的 Stream。

Streamfly 前身是離線作業(yè)平臺(tái) Omega 下的名為 Lambda 的子系統(tǒng)，它負(fù)責(zé)了所有實(shí)時(shí)作業(yè)的調(diào)度，最開始開始支持 Storm 和 Spark Streaming，后來改為只支持 Flink。在 2019 年的時(shí)候我們將 Lambda 獨(dú)立出來以此為基礎(chǔ)建立了 Streamfly 計(jì)算平臺(tái)。隨后，我們在 2019 年底開發(fā)并上線了第一個(gè)版本 Flink SQL 平臺(tái) StreamflySQL。這個(gè)版本基于模板 jar 提供了基本 Flink SQL 的功能，但是用戶體驗(yàn)還有待提升，因此我們在 2021 年年初從零開始重新建設(shè)了第二個(gè)版本的 StreamflySQL，而第二個(gè)版本是基于 SQL Gateway。

要了解這兩個(gè)版本的不同，我們需要先回顧下 Flink SQL 的基本工作流程。

用戶提交的 SQL 首先會(huì)被 Parser 解析為邏輯執(zhí)行計(jì)劃；邏輯執(zhí)行計(jì)劃經(jīng)過 Planner Optimizer 優(yōu)化，會(huì)生成物理執(zhí)行計(jì)劃；物理執(zhí)行計(jì)劃再通過 Planner CodeGen 代碼生成，翻譯為 DataStream API 常見的 Transformation；最后 StreamGraphGenerator 會(huì)將這些 Transformation 轉(zhuǎn)換為 Flink 作業(yè)的最終表示 JobGraph 提交到 Flink 集群。

上述一系列過程都發(fā)生在 TableEnvironment 里面。取決于部署模式的不同，TableEnvironment 可能運(yùn)行在 Flink Client 或者 JobManager 里。Flink 現(xiàn)在支持 3 種集群部署模式，包括 Application、 Per-Job 和 Session 模式。在 Application 模式下，TableEnvironment 會(huì)在 JobManager 端運(yùn)行，而在其余兩種模式下，TableEnvironment 都運(yùn)行在 Client 端。不過這三種模式都有一個(gè)共同的特點(diǎn)，TableEnvironment 都是一次性的，會(huì)在提交 JobGraph 之后自動(dòng)退出。

為了更好地復(fù)用 TableEnvironment 提高效率和提供有狀態(tài)的操作，有的項(xiàng)目會(huì)將 TableEnvironment 放到一個(gè)新的獨(dú)立 Server 端進(jìn)程里面去運(yùn)行，由此產(chǎn)生了一種新的架構(gòu)，我們稱之為 Server 端 SQL 編譯。相對地，還有 Client 端 SQL 編譯。

有同學(xué)可能會(huì)問，為什么沒有 JobManager 端 SQL 編譯，這是因?yàn)?JobManager 是相對封閉的組件，不適合拓展，而且即使做了達(dá)到的效果跟 Client 端編譯效果基本一樣。所以總體來看，一般就有 Client 和 Server 兩種常見的 Flink SQL 平臺(tái)架構(gòu)。

Client 端 SQL 編譯，顧名思義就是 SQL 的解析翻譯優(yōu)化都在 Client 端里進(jìn)行（這里的 Client 是廣義的 Client，并不一定是 Flink Client）。典型的案例就是通用模板 jar 和 Flink 的 SQL Client。這種架構(gòu)的優(yōu)點(diǎn)是開箱即用，開發(fā)成本低，而且使用的是 Flink public 的 API，版本升級(jí)比較容易；缺點(diǎn)是難以支持高級(jí)的功能，而且每次都要先啟動(dòng)一個(gè)比較重的 TableEnvironment 所以性能比較差。

然后是 Server 端 SQL 編輯。這種架構(gòu)將 SQL 解析翻譯優(yōu)化邏輯放到一個(gè)獨(dú)立的 Server 進(jìn)程去進(jìn)行，讓 Client 變得非常輕，比較接近于傳統(tǒng)數(shù)據(jù)庫的架構(gòu)。典型的案例是 Ververica 的 SQL Gateway。這種架構(gòu)的優(yōu)點(diǎn)是可拓展性好，可以支持很多定制化功能，而且性能好；缺點(diǎn)則是現(xiàn)在開源界沒有成熟的解決方案，像上面提到 SQL Gateway 只是一個(gè)比較初期的原型系統(tǒng)，缺乏很多企業(yè)級(jí)特性，如果用到生產(chǎn)環(huán)境需要經(jīng)過一定的改造，而且這些改造涉及比較多 Flink 內(nèi)部 API，需要比較多 Flink 的背景知識(shí)，總體來說開發(fā)成本比較高，而且后續(xù)版本升級(jí)工作量也比較大。

回到我們 Flink SQL 平臺(tái)，我們 StreamflySQL v1 是基于 Client 端 SQL 編譯，而 v2 是基于 Server 端的 SQL 編譯。下面就讓我逐個(gè)介紹一下。

02基于模板 jar 的 StreamflySQL v1

StreamflySQL v1 選擇 Client 端 SQL 編譯的主要原因有三個(gè):

• 首先是平臺(tái)集成。不同于很多公司的作業(yè)調(diào)度器用大數(shù)據(jù)中比較主流的 Java 編寫，我們的 Lambda 調(diào)度器是用 Go 開發(fā)的。這是因?yàn)?Lambda 在設(shè)計(jì)之初支持了多種實(shí)時(shí)計(jì)算框架，出于松耦合和公司技術(shù)棧的考慮，Lambda 以 Go 作為開發(fā)語言，會(huì)采用與 YARN 類似的動(dòng)態(tài)生成 Shell 腳本的方式來調(diào)用不同框架的命令行接口。這樣松耦合的接口方式給我們帶來很大的靈活性，比如我們可以輕松支持多個(gè)版本的 Flink，不需要強(qiáng)制用戶隨著系統(tǒng)版本升級(jí)，但同時(shí)也導(dǎo)致沒辦法直接去調(diào)用 Flink 原生的 Java API。
• 第二個(gè)原因是松耦合。開發(fā)的時(shí)候 Flink 版本是1.9，當(dāng)時(shí) Client API 比較復(fù)雜，不太適合平臺(tái)集成，并且當(dāng)時(shí)社區(qū)也在推動(dòng) Client 的重構(gòu)，所以我們盡量避免依賴 Client API去開發(fā) Flink SQL 平臺(tái)。
• 第三個(gè)原因是實(shí)踐經(jīng)驗(yàn)。因?yàn)槟０?jar + 配置中心模式在網(wǎng)易游戲內(nèi)部已經(jīng)有了比較多的應(yīng)用，所以我們在這方面積累了很多實(shí)踐經(jīng)驗(yàn)。綜合之下我們很自然地采用了模板 jar + 配置中心的架構(gòu)來實(shí)現(xiàn) v1 版本。

上圖是 v1 版本的整體架構(gòu)圖。我們在主要在 Lambda 作業(yè)平臺(tái)的基礎(chǔ)上新增了 StreamflySQL 后端作為配置中心，負(fù)責(zé)根據(jù)用戶提交的 SQL 和作業(yè)運(yùn)行配置加上通用的模板 jar 來生成一個(gè) Lambda 作業(yè)。

總體的作業(yè)提交流程如下:

1. 用戶在前端的 SQL 編輯器提交 SQL 和運(yùn)行配置。
2. StreamflySQL 后端收到請求后生成一個(gè) Lambda 作業(yè)并傳遞配置 ID。
3. 然后 Lambda 啟動(dòng)作業(yè)，背后是執(zhí)行 Flink CLI run 命令來提交作業(yè)。、
4. Flink CLI run 命令會(huì)啟動(dòng) Flink Client 來加載并執(zhí)行模版 jar 的 main 函數(shù)，這時(shí)會(huì)讀取 SQL 和配置，并初始化 TableEnvironment。
5. TableEnvironment 會(huì)從 Catalog 讀取必要的 Database/Table 等元信息。這里順帶一提是，在網(wǎng)易游戲我們沒有使用統(tǒng)一的 Catalog 來維護(hù)不同組件的元信息，而是不同組件有自己的元數(shù)據(jù)中心，對應(yīng)不同的 Catalog。
6. 最后 TableEnvironment 編譯好 JobGraph，以 Per-Job Cluster 的方式部署作業(yè)。

StreamflySQL v1 實(shí)現(xiàn)了 Flink SQL 平臺(tái)從零到一的建設(shè)，滿足了部分業(yè)務(wù)需求，但仍有不少痛點(diǎn)。

第一個(gè)痛點(diǎn)是響應(yīng)慢。

以一個(gè)比較典型的 SQL 來說，以模板 jar 的方式啟動(dòng)作業(yè)需要準(zhǔn)備 TableEnviroment，這可能會(huì)花費(fèi) 5 秒鐘，然后執(zhí)行 SQL 的編譯優(yōu)化包括與 Catalog 交互去獲取元數(shù)據(jù)，也可能會(huì)花費(fèi) 5 秒鐘；編譯得到j(luò)obgraph之后還需要準(zhǔn)備 per-job cluster，一般來說也會(huì)花費(fèi) 20 秒以上；最后還需要等待 Flink job的調(diào)度，也就是作業(yè)從 scheduled 變成 running 的狀態(tài)，這個(gè)可能也需要 10 秒鐘。

總體來說，v1 版本啟動(dòng)一個(gè) Flink SQL 作業(yè)至少需要 40 秒的時(shí)間，這樣的耗時(shí)相對來說是比較長的。但是仔細(xì)分析這些步驟，只有 SQL的編譯優(yōu)化和 job 調(diào)度是不可避免的，其他的比如 TableEnvironment 和 Flink cluster 其實(shí)都可以提前準(zhǔn)備，這里的慢就慢在資源是懶初始化的，而且?guī)缀鯖]有復(fù)用。

第二個(gè)痛點(diǎn)是調(diào)試難。

我們對 SQL 調(diào)試的需求有以下幾點(diǎn)：

• 第一點(diǎn)是調(diào)試的 SQL 與線上的 SQL 要基本一致。
• 第二點(diǎn)是調(diào)試 SQL 不能對線上的數(shù)據(jù)產(chǎn)生影響，它可以去讀線上的數(shù)據(jù)，但不能去寫。
• 第三點(diǎn)，因?yàn)檎{(diào)試的 SQL 通常只需要抽取少量的數(shù)據(jù)樣本就可以驗(yàn)證 SQL 的正確性，所以我們希望限制調(diào)試 SQL 的資源，一方面是出于成本的考慮，另外一方面也是為了防止調(diào)試的 SQL 與線上作業(yè)產(chǎn)生資源競爭。
• 第四點(diǎn)，因?yàn)檎{(diào)試 SQL 處理的數(shù)據(jù)量比較少，我們希望以更快更便捷的方式獲取到結(jié)果。

在 v1 版本中，我們對上述需求設(shè)計(jì)了如下解決方案：

1. 首先對于調(diào)試的 SQL，系統(tǒng)會(huì)在 SQL 翻譯的時(shí)候?qū)⒃瓉淼囊粋€(gè) Sink 替換為專用的 PrintSink，這解決了需求中的前兩點(diǎn)。
2. 然后對 PrintSink 進(jìn)行限流，通過 Flink 的反壓機(jī)制達(dá)到總體的限流，并且會(huì)限制作業(yè)的最長執(zhí)行時(shí)間，超時(shí)之后系統(tǒng)會(huì)自動(dòng)把作業(yè)結(jié)束掉，這解決了需求中的資源限制這點(diǎn)。
3. 最后為了更快地響應(yīng)，調(diào)試的作業(yè)并不會(huì)提交到 YARN 集群上去運(yùn)行，而是會(huì)在 Lamdba 服務(wù)器本地開啟開啟一個(gè) MiniCluster 去執(zhí)行，同時(shí)也方便我們從標(biāo)準(zhǔn)輸出去提取 PrintSink 的結(jié)果，這點(diǎn)解決了需求中的最后一點(diǎn)。

調(diào)試模式的架構(gòu)如上圖所示，比起一般的 SQL 提交流程，主要區(qū)別在于作業(yè)不會(huì)提交到 YARN 上，而是在 Lambda 服務(wù)器的本地執(zhí)行，從而節(jié)省了準(zhǔn)備 Flink 集群的開銷，并且更容易管控資源和獲取結(jié)果。

上述調(diào)試解決方案基本可用，但是實(shí)際使用過程中依然存在不少問題。

• 第一，如果用戶提交的 SQL 比較復(fù)雜，那么 SQL 的編譯優(yōu)化可能會(huì)耗費(fèi)比較久的時(shí)間，這會(huì)導(dǎo)致作業(yè)很容易超時(shí)，在有結(jié)果輸出之前可能就被系統(tǒng)結(jié)束掉，同時(shí)這樣的 SQL 也會(huì)給服務(wù)器造成比較大的壓力。
• 第二，該架構(gòu)沒法去調(diào)試時(shí)間窗口比較長的作業(yè)或者需要 Bootstrap State 的作業(yè)。
• 第三，因?yàn)閳?zhí)行結(jié)果是在作業(yè)結(jié)束之后才批量返回的，不是在作業(yè)執(zhí)行過程中就流式返回，因此用戶需要等到作業(yè)結(jié)束——通常是 10 分鐘以上才可以看到結(jié)果。
• 第四，在 SQL 的翻譯階段把調(diào)試 SQL 的 Sink 替換掉，這個(gè)功能是通過改造 Flink 的 Planner 來實(shí)現(xiàn)的，相當(dāng)于業(yè)務(wù)邏輯入侵到了 Planner 里面，這樣并不優(yōu)雅。

第三個(gè)痛點(diǎn)是 v1 版本只允許單條 DML。

相比傳統(tǒng)的數(shù)據(jù)庫，我們支持的 SQL 語句是很有限的，比如，MySQL 的 SQL 可以分成 DML、DQL、DDL 和 DCL。

• DML 用于操控?cái)?shù)據(jù)，常見的語句有 INSERT / UPDATE / DELETE。StreamflySQL v1 只支持了 INSERT，這和 Flink SQL 是保持一致的。Flink SQL 用 Retract 模式 — 也就是類似 Changelog 的方式 — 來表示 UPDATE/DELETE，所以只支持 INSERT，這點(diǎn)其實(shí)沒有問題。
• DQL 用于查詢數(shù)據(jù)，常見語句是 SELECT。這在 Flink SQL 是支持的，但因?yàn)槿狈?Sink 不能生成一個(gè)有意義的 Flink 作業(yè)，所以 StreamflySQL v1 不支持 DQL。
• DDL 用于定義元數(shù)據(jù)，常見語句是 CREATE / ALTER /DROP 等。這在 StreamflySQL v1 版本是不支持的，因?yàn)槟０?jar 調(diào)用 SQL 的入口是 sqlUpdate，不支持純元數(shù)據(jù)的操作，而且為純元數(shù)據(jù)的操作單獨(dú)啟動(dòng)一個(gè) TableEnvironment 來執(zhí)行也是完全不劃算。
• 最后是 DCL，用于管理數(shù)據(jù)權(quán)限，比如 GRANT 跟 REVOKE 語句。這個(gè) Flink SQL 是不支持的，原因是 Flink 目前只是數(shù)據(jù)的用戶而不是管理者，DCL 并沒有意義。

綜合來看，v1 版本只支持了單條 DML，這讓我們很漂亮的 SQL 編輯器變得空有其表?；谝陨线@些痛點(diǎn)，我們在今年調(diào)研并開發(fā)了 StreamflySQL v2。v2 采用的是 Server 端 SQL 編譯的架構(gòu)。

03基于 SQL Gateway 的 StreamflySQL v2

我們的核心需求是解決 v1 版本的幾個(gè)痛點(diǎn)，包括改善用戶體驗(yàn)和提供更完整的 SQL 支持?？傮w的思路是采用 Server 端的 SQL 編譯的架構(gòu)，提高可拓展性和性能。此外，我們的集群部署模式也改成 Session Cluster，預(yù)先準(zhǔn)備好集群資源，省去啟動(dòng) YARN application 的時(shí)間。

這里會(huì)有兩個(gè)關(guān)鍵問題。

• 首先是我們要完全自研還是基于開源項(xiàng)目？在調(diào)研期間我們發(fā)現(xiàn) Ververica 的 SQL Gateway 項(xiàng)目很符合我們需求，容易拓展而且是 Flink 社區(qū) FLIP-91 SQL Gateway 的一個(gè)基礎(chǔ)實(shí)現(xiàn)，后續(xù)也容易與社區(qū)的發(fā)展方向融合。
• 第二個(gè)問題是，SQL Gateway 本身有提交作業(yè)的能力，這點(diǎn)跟我們已有的 Lambda 平臺(tái)是重合的，會(huì)造成重復(fù)建設(shè)和難以統(tǒng)一管理的問題，比如認(rèn)證授權(quán)、資源管理、監(jiān)控告警等都會(huì)有兩個(gè)入口。那么兩者應(yīng)當(dāng)如何進(jìn)行分工？我們最終的解決方案是，利用 Session Cluster 的兩階段調(diào)度，即資源初始化和作業(yè)執(zhí)行是分離的，所以我們可以讓 Lambda 負(fù)責(zé) Session Cluster 的管理，而 StreamflySQL 負(fù)責(zé) SQL 作業(yè)的管理，這樣能復(fù)用 Lambda 大部分的基礎(chǔ)能力。

這是 StreamflySQL v2 的架構(gòu)圖。我們將 SQL Gateway 內(nèi)嵌到 SpringBoot 應(yīng)用中，開發(fā)了新的后端。總體看起來比 v1 版本要復(fù)雜，原因是原本的一級(jí)調(diào)度變成了會(huì)話和作業(yè)的兩級(jí)調(diào)度。

首先用戶需要?jiǎng)?chuàng)建一個(gè) SQL 會(huì)話，StreamflySQL 后端會(huì)生成一個(gè)會(huì)話作業(yè)。在 Lambda 看來會(huì)話作業(yè)是一種特殊作業(yè)，啟動(dòng)時(shí)會(huì)使用 yarn-session 的腳本來啟動(dòng)一個(gè) Flink Session Cluster。在 Session Cluster 初始化之后，用戶就可以在會(huì)話內(nèi)去提交 SQL。StreamflySQL 后端會(huì)給每個(gè)會(huì)話開啟一個(gè) TableEnvironment，負(fù)責(zé)執(zhí)行 SQL 語句。如果是只涉及元數(shù)據(jù)的 SQL，會(huì)直接調(diào)用 Catalog 接口完成，如果是作業(yè)類型的 SQL，會(huì)編譯成 JobGraph 提交到 Session Cluster 去執(zhí)行。

v2 版本很大程度上解決了 v1 版本的幾個(gè)痛點(diǎn):

• 在響應(yīng)時(shí)間方面，v1 常常會(huì)需要 1 分鐘左右，而 v2 版本通常在 10 秒內(nèi)完成。
• 在調(diào)試預(yù)覽方面，v2 不需要等作業(yè)結(jié)束，而是在作業(yè)運(yùn)行時(shí)，將結(jié)果通過 socket 流式地返回。這點(diǎn)是依賴了 SQL gateway 比較巧妙的設(shè)計(jì)。對于 select 語句，SQL Gateway 會(huì)自動(dòng)注冊一個(gè)基于 socket 的臨時(shí)表，并將 select 結(jié)果寫入到這個(gè)表。
• 在 SQL 支持方面，v1 只支持 DML，而 v2 借助于 SQL Gateway 可以支持 DML/DQL/DDL。

不過 SQL Gateway 雖然有不錯(cuò)的核心功能，但我們使用起來并不是一帆風(fēng)順，也遇到一些挑戰(zhàn)。

首先最為重要的是元數(shù)據(jù)的持久化。

SQL Gateway 本身的元數(shù)據(jù)只保存在內(nèi)存中，如果進(jìn)程重啟或是遇到異常崩潰，就會(huì)導(dǎo)致元數(shù)據(jù)丟失，這在企業(yè)的生產(chǎn)環(huán)境里面是不可接受的。因此我們將 SQL Gateway 集成到 SpringBoot 程序之后，很自然地就將元數(shù)據(jù)保存到了數(shù)據(jù)庫。

元數(shù)據(jù)主要是會(huì)話元數(shù)據(jù)，包括會(huì)話的 Catalog、Function、Table 和作業(yè)等等。這些元數(shù)據(jù)按照作用范圍可以分為 4 層。底下的兩層是全局的配置，以配置文件的形式存在；上面兩層是運(yùn)行時(shí)動(dòng)態(tài)生成的元數(shù)據(jù)，存在數(shù)據(jù)庫中。上層的配置項(xiàng)優(yōu)先級(jí)更高，可以用于覆蓋下層的配置。

我們從下往上看這些元數(shù)據(jù)：

• 最底層是全局的默認(rèn) Flink Configuration，也就是我們在 Flink Home 下的 flink-conf yaml 配置。
• 再上面一層是 Gateway 自身的配置，比如部署模式（比如是 YARN 還是 K8S），比如默認(rèn)要出冊的 Catalog 和 Function 等等。
• 第三層是 Session 會(huì)話級(jí)別的 Session Configuraion，比如會(huì)話對應(yīng)的 Session Cluster 的集群 ID 或者 TaskManager 的資源配置等等。
• 最上面一層是 Job 級(jí)別的配置，包括作業(yè)動(dòng)態(tài)生成的元數(shù)據(jù)，比如作業(yè) ID、用戶設(shè)置 checkpoint 周期等等。

這樣比較靈活的設(shè)計(jì)除了解決了元數(shù)據(jù)持久化的問題，也為我們的多租戶特性奠定了基礎(chǔ)。

第二個(gè)挑戰(zhàn)是多租戶。

多租戶分為資源和認(rèn)證兩個(gè)方面：

• 在資源方面，StreamflySQL 利用 Lambda 作業(yè)平臺(tái)可以在不同的隊(duì)列啟動(dòng) Session Cluster，它們的 Master 節(jié)點(diǎn)和資源很自然就是隔離的，所以沒有像 Spark Thrift Server 那樣不同用戶共用一個(gè) Master 節(jié)點(diǎn)和混用資源的問題。
• 在認(rèn)證方面，因?yàn)?Session Cluster 屬于不同用戶，所以 StreamflySQL 后端需要實(shí)現(xiàn)多租戶的偽裝。在網(wǎng)易游戲，組件一般會(huì)使用 Kerberos 認(rèn)證。我們采用多租戶實(shí)現(xiàn)的方式是使用 Hadoop 的 Proxy User，先登錄為超級(jí)用戶，然后偽裝成項(xiàng)目用戶來向不同組件獲取 delegation token，這里的組件主要是 Hive MetaStore 跟 HDFS，最后把這些 token 存到 UGI 里面并用 doAS 的方式來提交作業(yè)。

第三個(gè)挑戰(zhàn)是水平拓展。

為了高可用和拓展服務(wù)能力，StreamflySQL 很自然需要以多實(shí)例的架構(gòu)部署。因?yàn)槲覀円呀?jīng)將主要的狀態(tài)元數(shù)據(jù)存到數(shù)據(jù)庫，我們可以隨時(shí)從數(shù)據(jù)庫構(gòu)建出一個(gè)新的 TableEnvironment，所以 StreamflySQL 實(shí)例類似普通 Web 服務(wù)一樣非常輕，可以很容易地?cái)U(kuò)容縮容。

但是并不是所有狀態(tài)都可以持久化的，另外有些狀態(tài)我們故意會(huì)不持久化。比如用戶使用 SET 命令來改變 TableEnvironment 的屬性，比如開啟 Table Hints，這些屬于臨時(shí)屬性，會(huì)在重建 TableEnvironment 后被重置。這是符合預(yù)期的。再比如用戶提交 select 查詢做調(diào)試預(yù)覽時(shí)，TaskManager 會(huì)與 StreamflySQL 后端建立 socket 鏈接，而 socket 鏈接顯然也是不可持久化的。因此我們在 StreamflySQL 的多實(shí)例前加了親和性的負(fù)載均衡，按照 Session ID 來調(diào)度流量，讓在正常情況下同一個(gè)用戶的請求都落到同一個(gè)實(shí)例上，確保用戶使用體驗(yàn)的連續(xù)性。

第四個(gè)挑戰(zhàn)是作業(yè)狀態(tài)管理。

其實(shí)這里的狀態(tài)一詞是雙關(guān)，有兩個(gè)含義：

• 第一個(gè)含義是作業(yè)的運(yùn)行狀態(tài)。SQL gateway 目前只是提交 SQL 并不監(jiān)控后續(xù)的運(yùn)行狀態(tài)。因此，StreamflySQL 設(shè)置了監(jiān)控線程池來定時(shí)輪詢并更新作業(yè)狀態(tài)。因?yàn)?StreamflySQL 是多實(shí)例的，它們的監(jiān)控線程同時(shí)操作同一個(gè)作業(yè)的話，可能會(huì)有更新丟失的問題，所以我們這里使用了 CAS 樂觀鎖來保證過時(shí)的更新不會(huì)生效。然后我們會(huì)在作業(yè)異常退出或者無法獲取狀態(tài)時(shí)進(jìn)行告警，比如 JobManager 進(jìn)行 failover 的情況下，我們無法得知 Flink 作業(yè)的狀態(tài)，這時(shí)系統(tǒng)就會(huì)發(fā)出 disconnected 的異常狀態(tài)告警。
• 第二個(gè)含義是 Flink 的持久化狀態(tài)，即 Flink State。原生的 SQL gateway 并沒有管理 Flink 的 Savepoint 和 Checkpoint，因此我們加上了 stop 和 stop-with-savepoint 的功能，并強(qiáng)制開啟 retained checkpoint。這使得在作業(yè)遇到異常終止或者簡單 stop 之后，再次重啟時(shí)系統(tǒng)可以自動(dòng)查找到最新的 checkpoint。

這里我可以分享下我們的算法。其實(shí)自動(dòng)查找最新 checkpoint 的功能 Lambda 也有提供，但是 Lambda 假設(shè)作業(yè)都是 Per-Job Cluster，因此只要查找集群 checkpoint 目錄里最新的一個(gè) checkpoint 就可以了。但這樣的算法對 StreamflySQL 卻不適用，因?yàn)?Session Cluster 有多個(gè)作業(yè)，最新的 checkpoint 并不一定是我們目標(biāo)作業(yè)的。因此，我們改為了使用類似 JobManager HA 的查找方式，先讀取作業(yè)歸檔目錄元數(shù)據(jù)，從里面提取最新的一個(gè) checkpoint。

04未來工作

• 未來我們首先要解決的一個(gè)問題是 State 遷移的問題，即用戶對 SQL 進(jìn)行變更后，如何從原先的 Savepoint 進(jìn)行恢復(fù)。目前只能通過變更類型來告知用戶風(fēng)險(xiǎn)，比如通常而言加減字段不會(huì)造成 Savepoint 的不兼容，但如果新增一個(gè) join 表，造成的影響就很難說了。因此后續(xù)我們計(jì)劃通過分析 SQL 變更前后的執(zhí)行計(jì)劃，來預(yù)先告知用戶變更前后的狀態(tài)兼容性。
• 第二個(gè)問題是細(xì)粒度的資源管理。目前我們并不能在作業(yè)編譯時(shí)去指定 SQL 的資源，比如 TaskManager 的 CPU 和內(nèi)存在 Session Cluster 啟動(dòng)之后就確定了，是會(huì)話級(jí)別的。目前調(diào)整資源只能通過作業(yè)并行度調(diào)整，很不靈活并且容易造成浪費(fèi)。現(xiàn)在 Flink 1.14 已經(jīng)支持了 DataStream API 的細(xì)粒度資源管理，可以在算子級(jí)別設(shè)置資源，但 SQL API 現(xiàn)在還沒有計(jì)劃，后續(xù)我們可能參與進(jìn)去推動(dòng)相關(guān)議案的進(jìn)展。
• 最后是社區(qū)貢獻(xiàn)。我們對 SQL Gateway 有一定使用經(jīng)驗(yàn)，而且也對其進(jìn)行了不少的改進(jìn)，后續(xù)希望這些改進(jìn)能回饋給 Flink 社區(qū)，推動(dòng) FLIP-91 SQL Gateway 的進(jìn)展。?