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

背景

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

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

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

State Processor API 介紹

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

下面簡(jiǎn)單介紹一下如何使用 State Processor API 來(lái)完成 State 的查詢：

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

接下來(lái)為大家簡(jiǎn)述一下 State 查詢背后的原理。

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

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

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

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

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

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

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

在這個(gè)方案中，我們需要解決兩個(gè)問(wèn)題：

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

StateMeta Snapshot 機(jī)制

首先我們來(lái)回答第一個(gè)問(wèn)題，查詢一個(gè) State 需要哪些信息呢？

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

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

StateMeta Snapshot 簡(jiǎn)單來(lái)說(shuō)就是把狀態(tài)的元信息添加到 Savepoint Metadata 的過(guò)程，具體步驟如下：

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

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

State as Database

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

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

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

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

當(dāng)我們抽象出了單個(gè) State 之后，想要表示多個(gè) State 就比較容易了?？梢钥吹皆谏蠄D的例子中，這個(gè)算子包含 3 個(gè) State，分別是兩個(gè) KeyedState 和一個(gè) OperatorState，我們只需要將這些 Table 簡(jiǎn)單的 union 起來(lái)，再通過(guò) state_name 字段去區(qū)分不同的 State，就可以表示這個(gè)算子中所有的 State。

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

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

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

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

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

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

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

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

未來(lái)展望

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

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