字節跳動 Flink 狀態查詢實踐與優化

摘要：本文整理自字節跳動基礎架構工程師，Apache Flink Contributor 馬越在 Flink Forward Asia 2021 平臺建設專場的演講。主要內容包括：

  1. 背景

  2. State Processor API 介紹

  3. StateMeta Snapshot 機制

  4. State as Database

  5. 使用 Flink Batch SQL 查詢任務狀態

  6. 未來展望

01 背景

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

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

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

02 State Processor API 介紹

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

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

  1. 首先創建 ExistingSavepoint 用來表示一個 Savepoint。初始化 ExistingSavepoint 時需要提供 Savepoint 路徑和 StateBackend 等信息；

  2. 然後實現 ReaderFunction 用於重新註冊所需要查詢的  State 以及定義處理 State 的方式。查詢狀態的過程中會遍歷所有的 Key 並按照我們定義的方式去操作 State；

  3. 最後，調用 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 時能用最簡單的方式；同時也不需要知道任何信息。

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

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

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

03 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 的過程，具體步驟如下：

  1. 首先在 State 註冊的時候，Task 會把 operatorName\ID\KeySerializer\StateDescriptors 等元信息都保存在 Task 的內存中；

  2. 觸發 Savepoint 時，Task 會在製作快照的同時，對狀態的元信息也同樣進行快照。快照完成之後將狀態的元信息 (StateMeta) 和狀態文件的句柄 (StateHandle) 一起上報給 JobManager；

  3. JobManager 在收到所有 Task 上報的 StateMeta 信息之後 ，將這些狀態元信息進行合併，最後會把合併之後的狀態元信息保存到 Savepoint 目錄里名爲 stateInfo 的文件中。

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

04 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。

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

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

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

05 使用 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。用戶使用這樣的方式就可以比較方便地驗證狀態是否正確。

06 未來展望

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

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

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/GT1OWi__HTfundXVGt4XkQ

猜你喜歡