有狀态的計算作為容錯以及資料一緻性的保證，是當今實時計算必不可少的特性之一，流行的實時計算引擎包括 Google Dataflow、Flink、Spark (Structure) Streaming、Kafka Streams 都分别提供對内置 State 的支援。State 的引入使得實時應用可以不依賴外部資料庫來存儲中繼資料及中間資料，部分情況下甚至可以直接用 State 存儲結果資料，這讓業界不禁思考: State 和 Database 是何種關系？有沒有可能用 State 來代替資料庫呢？
在這個課題上，Flink 社群是比較早就開始探索的。總體來說，Flink 社群的努力可以分為兩條線: 一是在作業運作時通過作業查詢接口通路 State 的能力，即 QueryableState；二是通過 State 的離線 dump 檔案（Savepoint）來離線查詢和修改 State 的能力，即即将引入的 Savepoint Processor API。
QueryableState
在 2017 年釋出的 Flink 1.2 版本，Flink 引入了 QueryableState 的特性以允許使用者通過特定的 client 查詢作業 State 的内容 [1]，這意味着 Flink 應用可以在完全不依賴 State 存儲媒體以外的外部存儲的情況下提供實時通路計算結果的能力。
隻通過 Queryable State 提供實時資料通路
然而，QueryableState 雖然設想上比較理想化，但由于依賴底層架構的改動較多且功能也比較受限，它一直處于 Beta 版本并不能用于生産環境。針對這個問題，在前段時間騰訊的工程師楊華提出 QueryableState 的改進計劃 [2]。在郵件清單中，社群就 QueryableState 是否可以用于代替資料庫作了讨論并出現了不同的觀點。筆者結合個人見解将 State as Database 的主要優缺點整理如下。
優點:
- 更低的資料延遲。一般情況下 Flink 應用的計算結果需要同步到外部的資料庫，比如定時觸發輸出視窗計算結果，而這種同步通常是定時的會帶來一定的延遲，導緻計算是實時的而查詢卻不是實時的尴尬局面，而直接 State 則可以避免這個問題。
- 更強的資料一緻性保證。根據外部存儲的特性不同，Flink Connector 或者自定義的 SinkFunction 提供的一緻性保障也有所差别。比如對于不支援多行事務的 HBase，Flink 隻能通過業務邏輯的幂等性來保障 Exactly-Once 投遞。相比之下 State 則有妥妥的 Exactly-Once 投遞保證。
- 節省資源。因為減少了同步資料到外部存儲的需要，我們可以節省序列化和網絡傳輸的成本，另外當然還可以節省資料庫成本。
缺點:
- SLA 保障不足。資料庫技術已經非常成熟，在可用性、容錯性和運維上都很多的積累，在這點上 State 還相當于是處于原始人時期。另外從定位上來看，Flink 作業有版本疊代維護或者遇到錯誤自動重新開機帶來的 down time，并不能達到資料庫在資料通路上的高可用性。
- 可能導緻作業的不穩定。未經過考慮的 Ad-hoc Query 可能會要求掃描并傳回誇張量級的資料，這會系統帶來很大的負荷，很可能影響作業的正常執行。即使是合理的 Query，在并發數較多的情況下也可能影響作業的執行效率。
- 存儲資料量不能太大。State 運作時主要存儲在 TaskManager 本地記憶體和磁盤，State 過大會造成 TaskManager OOM 或者磁盤空間不足。另外 State 大意味着 checkpoint 大，導緻 checkpoint 可能會逾時并顯著延長作業恢複時長。
- 隻支援最基礎的查詢。State 隻能進行最簡單的資料結構查詢，不能像關系型資料庫一樣提供函數等計算能力，也不支援謂詞下推等優化技術。
- 隻可以讀取，不能修改。State 在運作時隻可以被作業本身修改，如果實在要修改 State 隻能通過下文的 Savepoint Processor API 來實作。
總體來說，目前 State 代替資料庫的缺點還是遠多于其優點，不過對于某些對資料可用性要求不高的作業來說，使用 State 作為資料庫還是完全合理的。由于定位上的不同，Flink State 在短時間内很難看到可以完全替代資料庫的可能性，但在資料通路特性上 State 往資料庫方向發展是無需質疑的。
Savepoint Processor API
Savepoint Processor API 是社群最近提出的一個新特性（見 FLIP-42 [3]），用于離線對 State 的 dump 檔案 Savepoint 進行分析、修改或者直接根據資料建構出一個初始的 Savepoint。Savepoint Processor API 屬于 Flink State Evolution 的 State Management。如果說 QueryableState 是 DSL 的話，Flink State Evolution 就是 DML，而 Savepoint Processor API 就是 DML 中最為重要的部分。
Savepoint Processor API 的前身是第三方的 Bravo 項目 [4]，主要思路提供 Savepoint 和 DataSet 互相轉換的能力，典型應用是 Savepoint 讀取成 DataSet，在 DataSet 上進行修改，然後再寫為一個新的 Savepoint。這适合用于以下的場景:
- 分析作業 State 以研究其模式和規律
- 排查問題或者審計
- 為新的應用建構的初始 State
- 修改 Savepoint，比如:
- 改變作業最大并行度
- 進行巨大的 Schema 改動
- 修正有問題的 State
Savepoint 作為 State 的 dump 檔案，通過 Savepoint Processor API 可以暴露資料查詢和修改功能，類似于一個離線的資料庫，但 State 的概念和典型關系型資料的概念還是有很多不同，FLIP-43 也對這些差異進行了類比和總結。
首先 Savepoint 是多個 operator 的 state 的實體存儲集合，不同 operator 的 state 是獨立的，這類似于資料庫下不同 namespace 之間的 table。我們可以得到 Savepoint 對應資料庫，單個 operator 對應 Namespace。
| Database | Savepoint |
|---|---|
| Namespace | Uid |
| Table | State |
但就 table 而言，其在 Savepoint 裡對應的概念根據 State 類型的不同而有所差别。State 有 Operator State、Keyed State 和 Broadcast State 三種，其中 Operator State 和 Broadcast State 屬于 non-partitioned state，即沒有按 key 分區的 state，而相反地 Keyed State 則屬于 partitioned state。對于 non-partitioned state 來說，state 是一個 table，state 的每個元素即是 table 裡的一行；而對于 partitioned state 來說，同一個 operator 下的所有 state 對應一個 table。這個 table 像是 HBase 一樣有個 row key，然後每個具體的 state 對應 table 裡的一個 column。
舉個例子，假設有一個遊戲玩家得分和線上時長的資料流，我們需要用 Keyed State 來記錄玩家所在組的分數和遊戲時長，用 Operator State 記錄玩家的總得分和總時長。
在一段時間内資料流的輸入如下:
| user_id | user_name | user_group | score |
|---|---|---|---|
| 1001 | Paul | A | 5,000 |
| 1002 | Charlotte | 3,600 | |
| 1003 | Kate | C | 2,000 |
| 1004 | Robert | B | 3,900 |
| time | |||
|---|---|---|---|
| 1,800 | |||
| 1,200 | |||
| 600 | |||
用 Keyed State ，我們分别注冊 group_score 和 group_time 兩個 MapState 表示組總得分群組總時長，并根據 user_group keyby 資料流之後将兩個名額的累積值更新到 State 裡，得到的表如下:
| group_score | group_time | |
|---|---|---|
| 8,600 | 3,000 | |
| 2,00 | ||
相對地，假如用 Operator State 來記錄總得分和總時長（并行度設為 1），我們注冊 total_score 和 total_time 兩個 State，得到的表有兩個:
total_score |
------- |
14,500 |
| total_time |
|---|
| 5,600 |
至此 Savepoint 和 Database 的對應關系應該是比較清晰明了的。而對于 Savepoint 來說還有不同的 StateBackend 來決定 State 具體如何持續化，這顯然對應的是資料庫的存儲引擎。在 MySQL 中，我們可以通過簡單的一行指令 ALTER TABLE xxx ENGINE = InnoDB; 來改變存儲引擎，在背後 MySQL 會自動完成繁瑣的格式轉換工作。而對于 Savepoint 來說，由于 StateBackend 各自的存儲格式不相容，目前尚不能友善地切換 StateBackend。為此，社群在不久前建立 FLIP-41 [5] 來進一步完善 Savepoint 的可操作性。
總結
State as Database 是實時計算發展的大趨勢，它并不是要代替資料庫的使用，而是借鑒資料庫領域的經驗拓展 State 接口使其操作方式更接近我們熟悉的資料庫。對于 Flink 而言，State 的外部使用可以分為線上的實時通路和離線的通路和修改，分别将由 Queryable State 和 Savepoint Processor API 兩個特性支援。
參考文獻
- Queryable State in Apache Flink® 1.2.0: An Overview & Demo
- Improve Queryable State and Introduce a QueryServerProxy Component
- FLIP-43: Savepoint Processor API
- Bravo: Utilities for processing Flink checkpoints/savepoints
- FLIP-41: Unify Keyed State Snapshot Binary Format for Savepoints
作者介紹：
林小鉑，網易遊戲進階開發工程師，負責遊戲資料中心實時平台的開發及運維工作，目前專注于 Apache Flink 的開發及應用。探究問題本來就是一種樂趣。
原文連結：
http://www.whitewood.me/2019/06/02/Flink-State-As-Database/Flink Forward Asia 倒計時 16 天！11 月 28-30 日，Flink Forward Asia 2019 核心技術專場，屆時 Apache Flink 核心貢獻者們将與多位來自一線的業界資深專家帶你全方位解鎖 Flink 核心技術。購票及了解更多大會詳情，可點選「
大會詳情」。
（11月28日下午，專場議程）
（11月29日上午，專場議程）