本篇文章介紹了字節跳動在 Flink 狀態查詢方面所進行的優化，解決了查詢 Flink 任務狀態時開發成本高及無法查詢狀態元信息等問題，提出了 State Query on Flink SQL 的解決方案，讓用戶使用 Flink Batch SQL 就可以快速查詢 Flink 任務狀態。

背景

眾所周知，Flink 中的 State 保存了算子計算過程的中間結果。當任務出現異常時，可以通過查詢任務快照中的 State 獲取有效線索。

但目前對于 Flink SQL 任務來說，當我們想要查詢作業 State 時，通常會因為無法獲知 State 的定義方式和具體類型等信息，而導致查詢 State 的成本過高。

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

State Processor API 介紹

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

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

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

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

在 Savepoint 目錄中包含兩種文件，一種是狀態數據文件，比如上圖中的 opA-1-state ，這個文件里面保存著算子 A 在第一個 SubTask 狀態的明細數據；還有一種元數據文件，對應上圖中的 _metadata，元數據文件中保存了每個算子和狀態文件的映射關系。

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

有些同學可能會問，既然社區已經提供了查詢 State 的功能，我們為什么還要去做同樣的工作呢？主要是因為我們在使用 State Processor API 的過程中發現一些問題：

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

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

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

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

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

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

StateMeta Snapshot 機制

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

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

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

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

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

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

State as Database

接下來我們來回答第二個問題，我們如何用 SQL 來表達 State。其實社區在設計 State Processor API 的時候就提出了一些解決思路，也就是 State As Database。

在傳統的數據庫中，通常用 Catalog、Database、Table 這個三個元素來表示一個 Table，其實我們也可以將用樣的邏輯到映射到 Flink State 上。我們可以把 Flink 的 State 當作一種特殊的數據源，作業每次產生的 Savepoint 都當作一個獨立 DB 。在這個 DB 中，我們將 State 元信息、State 的明細數據，都抽象成不同的 Table 暴露給用戶，用戶直接查詢這些 Table 就可以獲取任務的狀態信息。

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

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

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

最后還有一個問題，我們如何知道一個任務到底用了哪些 State 或者這些 State 的具體類型呢？

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

使用 Flink Batch SQL 查詢任務狀態

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

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

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

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

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

未來展望

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

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