本篇文章介紹了字節(jié)跳動在 Flink 狀態(tài)查詢方面所進(jìn)行的優(yōu)化，解決了查詢 Flink 任務(wù)狀態(tài)時開發(fā)成本高及無法查詢狀態(tài)元信息等問題，提出了 State Query on Flink SQL 的解決方案，讓用戶使用 Flink Batch SQL 就可以快速查詢 Flink 任務(wù)狀態(tài)。

背景

眾所周知，F(xiàn)link 中的 State 保存了算子計算過程的中間結(jié)果。當(dāng)任務(wù)出現(xiàn)異常時，可以通過查詢?nèi)蝿?wù)快照中的 State 獲取有效線索。

但目前對于 Flink SQL 任務(wù)來說，當(dāng)我們想要查詢作業(yè) State 時，通常會因?yàn)闊o法獲知 State 的定義方式和具體類型等信息，而導(dǎo)致查詢 State 的成本過高。

為了解決這個問題，字節(jié)跳動流式計算團(tuán)隊在內(nèi)部提出了 State Query on Flink SQL 的解決方案——用戶通過寫 SQL 的方式就可以簡單地查詢 State。本文將主要介紹字節(jié)跳動在 Flink 狀態(tài)查詢這方面所進(jìn)行的相關(guān)工作。

State Processor API 介紹

提到狀態(tài)查詢，我們自然會聯(lián)想到 Flink 在 1.9 版本提出的特性 -- State Processor API。使用 State Processor API，我們可以將作業(yè)產(chǎn)生的 Savepoint 轉(zhuǎn)換成 DataSet，然后使用 DataSet API 完成對 State 的查詢、修改和初始化等操作。

下面簡單介紹一下如何使用 State Processor API 來完成 State 的查詢：

• 首先創(chuàng)建 ExistingSavepoint 用來表示一個 Savepoint。初始化 ExistingSavepoint 時需要提供 Savepoint 路徑和 StateBackend 等信息；
• 然后實(shí)現(xiàn) ReaderFunction 用于重新注冊所需要查詢的 State 以及定義處理 State 的方式。查詢狀態(tài)的過程中會遍歷所有的 Key 并按照我們定義的方式去操作 State；
• 最后，調(diào)用 Savepoint.readKeyedState 并傳入算子的 uid 和 ReaderFunction，就可以完成 State 的查詢。

接下來為大家簡述一下 State 查詢背后的原理。

在 Savepoint 目錄中包含兩種文件，一種是狀態(tài)數(shù)據(jù)文件，比如上圖中的 opA-1-state ，這個文件里面保存著算子 A 在第一個 SubTask 狀態(tài)的明細(xì)數(shù)據(jù)；還有一種元數(shù)據(jù)文件，對應(yīng)上圖中的 _metadata，元數(shù)據(jù)文件中保存了每個算子和狀態(tài)文件的映射關(guān)系。

當(dāng)我們在進(jìn)行狀態(tài)查詢的時候。首先在 Client 端會根據(jù) Savepoint 路徑去解析 metadata 文件。通過算子 ID，可以獲取需要查詢的狀態(tài)所對應(yīng)的文件的句柄。當(dāng)狀態(tài)查詢真正執(zhí)行時，負(fù)責(zé)讀取狀態(tài)的 Task 會創(chuàng)建一個新的 StateBackend ，然后將狀態(tài)文件中的數(shù)據(jù)恢復(fù)到 Statebackend 中。等到狀態(tài)恢復(fù)完成之后就會遍歷全部的 Key 并把對應(yīng)的狀態(tài)交給 ReaderFunction 處理。

有些同學(xué)可能會問，既然社區(qū)已經(jīng)提供了查詢 State 的功能，我們?yōu)槭裁催€要去做同樣的工作呢？主要是因?yàn)槲覀冊谑褂?State Processor API 的過程中發(fā)現(xiàn)一些問題：

• 每次查詢 State 我們都需要獨(dú)立開發(fā)一個 Flink Batch 任務(wù)，對用戶來說具有一定的開發(fā)成本；
• 實(shí)現(xiàn) ReaderFunction 的時候需要比較清晰地了解任務(wù)狀態(tài)的定義方式，包括 State 的名稱、類型以及 State Descriptor 等信息，對用戶來說使用門檻高較高；
• 使用 State Processor API 時，只能查詢單個算子狀態(tài)，無法同時查詢多個算子的狀態(tài)；
• 無法直接查詢?nèi)蝿?wù)狀態(tài)的元信息，比如查詢?nèi)蝿?wù)使用了哪些狀態(tài)，或者查詢某個狀態(tài)的類型。

總體來說，我們的目標(biāo)有兩個，一是降低用戶的使用成本；二是增強(qiáng)狀態(tài)查詢的功能。我們希望用戶在查詢 State 時能用最簡單的方式；同時也不需要知道任何信息。

此外，我們還希望用戶能同時查詢多個算子的 State ，也可以直接查詢作業(yè)使用了哪些 State，每個 State 的類型是什么。

因此，我們提出了 State Query on Flink SQL 的解決方案。簡單來說是把 State 當(dāng)成數(shù)據(jù)庫一樣，讓用戶通過寫 SQL 的方式就可以很簡單地查詢 State。

在這個方案中，我們需要解決兩個問題：

• 如何對用戶屏蔽 State 的信息：參考 State Processor API 我們可以知道，查詢 State 需要提供非常多的信息，比如 Savepoint 路徑、 StateBacked 類型、算子 id 、State Descriptor 等等。通過 SQL 語句顯然難以完整地表述這些復(fù)雜的信息，那么查詢狀態(tài)到底需要哪些內(nèi)容，我們又如何對用戶屏蔽 State 里復(fù)雜的細(xì)節(jié)呢？這是我們面對的第一個難點(diǎn)。
• 如何用 SQL 表達(dá) State：State 在 Flink 中的存儲方式并不像 Database 一樣，我們?nèi)绾稳ビ?SQL 來表達(dá)狀態(tài)的查詢過程呢？這是我們要解決的另一個難點(diǎn)。

StateMeta Snapshot 機(jī)制

首先我們來回答第一個問題，查詢一個 State 需要哪些信息呢？

可以參考上文中 State Processor API 的示例，當(dāng)我們創(chuàng)建 ExistingSavepoint 和 ReaderFunction 的時候，我們需要提供的信息有 Savepoint 路徑、Backend 類型、OperatorID、算子 key 的類型、State 名稱以及 Serializer 等等，我們可以將這些統(tǒng)一稱為狀態(tài)的元信息。

對于 Flink SQL 任務(wù)來說，要清楚地了解這些信息，對用戶來說門檻是非常高的。我們的想法是讓用戶只需要提供最簡單的信息，即 Savepoint ID ，然后由 Flink 框架把其他的元信息都存在 Savepoint 中，這樣就可以對用戶屏蔽 State 那些復(fù)雜的細(xì)節(jié)，完成狀態(tài)的查詢。因此，我們引入了 StateMeta Snapshot 機(jī)制。

StateMeta Snapshot 簡單來說就是把狀態(tài)的元信息添加到 Savepoint Metadata 的過程，具體步驟如下：

• 首先在 State 注冊的時候，Task 會把 operatorName\ID\KeySerializer\StateDescriptors 等元信息都保存在 Task 的內(nèi)存中；
• 觸發(fā) Savepoint 時，Task 會在制作快照的同時，對狀態(tài)的元信息也同樣進(jìn)行快照。快照完成之后將狀態(tài)的元信息 (StateMeta) 和狀態(tài)文件的句柄 (StateHandle) 一起上報給 JobManager；
• JobManager 在收到所有 Task 上報的 StateMeta 信息之后 ，將這些狀態(tài)元信息進(jìn)行合并，最后會把合并之后的狀態(tài)元信息保存到 Savepoint 目錄里名為 stateInfo 的文件中。

之后在狀態(tài)查詢時就只需解析 Savepoint 中的 stateInfo 文件，而不再需要用戶通過代碼去輸入這些 State 的元信息。通過這樣的方式可以很大程度地降低用戶查詢狀態(tài)的成本。

State as Database

接下來我們來回答第二個問題，我們?nèi)绾斡?SQL 來表達(dá) State。其實(shí)社區(qū)在設(shè)計 State Processor API 的時候就提出了一些解決思路，也就是 State As Database。

在傳統(tǒng)的數(shù)據(jù)庫中，通常用 Catalog、Database、Table 這個三個元素來表示一個 Table，其實(shí)我們也可以將用樣的邏輯到映射到 Flink State 上。我們可以把 Flink 的 State 當(dāng)作一種特殊的數(shù)據(jù)源，作業(yè)每次產(chǎn)生的 Savepoint 都當(dāng)作一個獨(dú)立 DB 。在這個 DB 中，我們將 State 元信息、State 的明細(xì)數(shù)據(jù)，都抽象成不同的 Table 暴露給用戶，用戶直接查詢這些 Table 就可以獲取任務(wù)的狀態(tài)信息。

首先我們來看如何把 State 表示為 Table。我們都知道在 Flink 中，常用的 State 有兩種類型，分別是 KeyedState 和 OperatorState。

• 對于 OperatorState 來說，它只有 Value 這一個屬性，用來表示這個 State 具體的值。因此我們可以把 OperatorState 表示為只包含一個 Value 字段的表結(jié)構(gòu)。
• 對于 KeyedState 來說，每個 State 在不同的 Key 和 Namespace 下的值可能都不一樣， 因此我們可以將 KeyedState 表示為一個包含 Key、Namespace、Value 這三個字段的表結(jié)構(gòu)。

當(dāng)我們抽象出了單個 State 之后，想要表示多個 State 就比較容易了。可以看到在上圖的例子中，這個算子包含 3 個 State，分別是兩個 KeyedState 和一個 OperatorState，我們只需要將這些 Table 簡單的 union 起來，再通過 state_name 字段去區(qū)分不同的 State，就可以表示這個算子中所有的 State。

最后還有一個問題，我們?nèi)绾沃酪粋€任務(wù)到底用了哪些 State 或者這些 State 的具體類型呢？

為了解決這個問題，我們定義了一種特殊表 -- StateMeta ，用來表示一個 Flink 任務(wù)中所有 State 的元信息。StateMeta 中包含一個任務(wù)中每個 State 的名稱、State 所在的算子 ID 、算子名稱 、Key 的類型和 Value 的類型等等，這樣用戶直接查詢 StateMeta 這個表就能獲取任務(wù)中所有狀態(tài)的元信息。

使用 Flink Batch SQL 查詢?nèi)蝿?wù)狀態(tài)

以上就是狀態(tài)查詢方案的整體介紹。那我們到底如何去查詢一個 State 呢，我們以一個 Word Count 任務(wù)為例來說明。

首先，我們需要創(chuàng)建一個 Flink SQL 任務(wù)并啟動。通過 web-ui 可以看到這個任務(wù)中包含三個算子，分別是 Source，Aggregate 還有 Sink。然后，我們可以觸發(fā) Savepoint，當(dāng) Savepoint 制作成功之后獲取對應(yīng)的 SavepointID。我們可以通過 SavepointID 去完成作業(yè)狀態(tài)的查詢。

假如我們現(xiàn)在對 Flink SQL 任務(wù)中狀態(tài)的使用一無所知，那么首先我們需要查詢的就是這個 Flink 任務(wù)中包含哪些 State 以及這些 State 的類型。我們可以從 StateMeta 表獲取這些信息。如上圖中場景一所示，通過查詢 StateMeta 表，可以看到這個任務(wù)包含一個 ListState 和一個 ValueState，分別存在于 Source 算子和 Aggregate 算子中。

此外，有些對 Flink 比較了解的同學(xué)知道，KafkaSource 中的 State 是用于記錄當(dāng)前消費(fèi)的 Offset 信息。如場景二所示，我們可以通過查詢 Source 算子的狀態(tài)，獲取到任務(wù)中消費(fèi) Kafka Topic 的 Partition 和 Offset 信息。

還有一種比較常見的場景，比如下游的業(yè)務(wù)同學(xué)發(fā)現(xiàn)某個 key（比如 key_662）的結(jié)果異常。我們在定位問題的時候可以直接去查詢作業(yè)中 aggregate 算子中的狀態(tài)，同時去指定 key 等于 key_662 作為查詢條件。如上圖場景三所示，通過查詢的結(jié)果可以看到，當(dāng) key 為 662 時對應(yīng)的聚合結(jié)果是 11290。用戶使用這樣的方式就可以比較方便地驗(yàn)證狀態(tài)是否正確。

未來展望

未來，我們計劃進(jìn)一步豐富 State 的功能，目前我們支持了使用 SQL 查詢 State 的功能 ，其實(shí)社區(qū)還提供了 State 修改和初始化的能力。在一些場景下，這些能力也比較重要。比如，我們已知狀態(tài)中的部分 key 計算錯誤，希望將狀態(tài)中這部分的數(shù)據(jù)進(jìn)行修正；或者任務(wù)邏輯發(fā)生變更以后和之前的狀態(tài)不能完全兼容， 這個時候我們希望可以通過狀態(tài)修改和初始化的能力去生成一個新的 Savepoint。同樣，在使用方式上我們也希望用戶能直接使用 SQL 中 insert 和 update 語法來完成狀態(tài)的修改和初始化操作。

其次，我們會進(jìn)一步加強(qiáng) State 的可用性。我們使用 DAG 編輯的方案解決了作業(yè)拓?fù)浒l(fā)生變化時產(chǎn)生的狀態(tài)不兼容問題，但是當(dāng) Flink SQL 任務(wù)修改字段時 State Serializer 可能會變化，同樣導(dǎo)致狀態(tài)無法兼容。針對這種情況我們設(shè)計了完整的 Flink SQL State Schema Evolution 方案，可以極大的增強(qiáng) Flink SQL 任務(wù)發(fā)生變化之后狀態(tài)的恢復(fù)能力，目前正在落地中。我們還提供了完善的狀態(tài)恢復(fù)事前檢查能力，能夠做到在任務(wù)上線之前就檢查出狀態(tài)是否兼容并告知用戶，避免狀態(tài)不兼容引起作業(yè)啟動失敗對線上造成影響。