1.什么是狀態(tài)

官方定義：當前計算流程需要依賴到之前計算的結果，那么之前計算的結果就是狀態(tài)。

這句話還是挺好理解的，狀態(tài)不只存在于Flink，也存在生活的方方面面，比如看到一個認識的人，如何識別認識呢？就是眼睛看到這個人的樣子，再和大腦記憶中的人做對比，就知道認識這個人，其中大腦記憶中的人就是存儲在狀態(tài)中。

狀態(tài)又分為無狀態(tài)和有狀態(tài)。

• 無狀態(tài)：例如消費延遲計算，單條輸入包含所有的信息，不依賴于歷史消息。在這種模式的計算中，無論這條輸入進來多少次，輸出的結果都是一樣的，因為單條輸入中已經包含了所需的所有信息。消費落后等于生產者減去消費者。生產者的消費在單條數(shù)據中可以得到，消費者的數(shù)據也可以在單條數(shù)據中得到，所以相同輸入可以得到相同輸出，這就是一個無狀態(tài)的計算。
• 有狀態(tài)：例如訪問量統(tǒng)計，單條輸入僅包含部分信息，依賴歷史消息。這種模式是將數(shù)據輸入算子中，用來進行各種復雜的計算并輸出數(shù)據。這個過程中算子會去訪問之前存儲在里面的狀態(tài)。另外一方面，它還會把現(xiàn)在的數(shù)據對狀態(tài)的影響實時更新，如果輸入100 條數(shù)據，最后輸出就是 100 條結果。

2.狀態(tài)應用場景

通常以下4種場景會用到狀態(tài)：

• 去重：比如上游的系統(tǒng)數(shù)據可能會有重復，落到下游系統(tǒng)時希望把重復的數(shù)據都去掉。去重需要先了解哪些數(shù)據來過，哪些數(shù)據還沒有來，也就是把所有的主鍵都記錄下來，當一條數(shù)據到來后，能夠看到在主鍵當中是否存在。
• 窗口計算：比如統(tǒng)計每分鐘 Nginx 日志 API 被訪問了多少次。窗口是一分鐘計算一次，在窗口觸發(fā)前，如 08:00 ~ 08:01 這個窗口，前59秒的數(shù)據來了需要先放入內存，即需要把這個窗口之內的數(shù)據先保留下來，等到 8:01 時一分鐘后，再將整個窗口內觸發(fā)的數(shù)據輸出。未觸發(fā)的窗口數(shù)據也是一種狀態(tài)。
• 機器學習/深度學習：如訓練的模型以及當前模型的參數(shù)也是一種狀態(tài)，機器學習可能每次都用有一個數(shù)據集，需要在數(shù)據集上進行學習，對模型進行一個反饋。
• 訪問歷史數(shù)據：比如與昨天的數(shù)據進行對比，需要訪問一些歷史數(shù)據。如果每次從外部去讀，對資源的消耗可能比較大，所以也希望把這些歷史數(shù)據也放入狀態(tài)中做對比。

3.狀態(tài)管理

實時計算中的狀態(tài)的功能主要體現(xiàn)在任務可以做到失敗重啟后沒有數(shù)據質量、時效問題。

• 數(shù)據質量問題：當實時任務掛掉后，從消息失敗offset位置開始消費，數(shù)據就錯誤。
• 數(shù)據時效問題：實時任務要求有時效性，當從源offset開始位置運行時，需要好幾個小時才能追上當前offset。時效性就很差。

針對以上問題，就引出了狀態(tài)管理。

當我們把數(shù)據定期（例如每隔10min）的給存儲到 HDFS 上面時，任務掛了、恢復之后。我們的任務還可以從 HDFS 上面把這個數(shù)據給讀回來，接著從最新的一個 Kafka Offset 繼續(xù)計算就可以，這樣既沒沒有數(shù)據質量問題，也沒有數(shù)據時效性問題。

因此，實時計算中提到的狀態(tài)的概念重點不止在于狀態(tài)本身，更重要的在于強調 "管理" 狀態(tài)。

基于上述，狀態(tài)管理對流式作業(yè)的要求總結如下：

• 7*24小時運行，高可靠；
• 數(shù)據不丟不重，恰好計算一次；
• 數(shù)據實時產出，不延遲；

但是基于以上要求，內存的管理就會出現(xiàn)一些問題。由于內存的容量是有限制的。如果要做 24 小時的窗口計算，將 24 小時的數(shù)據都放到內存，可能會出現(xiàn)內存不足；另外，作業(yè)是 7*24，需要保障高可用，機器若出現(xiàn)故障或者宕機，需要考慮如何備份及從備份中去恢復，保證運行的作業(yè)不受影響；此外，考慮橫向擴展，假如網站的訪問量不高，統(tǒng)計每個 API 訪問次數(shù)的程序可以用單線程去運行，但如果網站訪問量突然增加，單節(jié)點無法處理全部訪問數(shù)據，此時需要增加幾個節(jié)點進行橫向擴展，這時數(shù)據的狀態(tài)如何平均分配到新增加的節(jié)點也問題之一。因此，將數(shù)據都放到內存中，并不是最合適的一種狀態(tài)管理方式。

最理想的狀態(tài)管理需要滿足易用、高效、可靠三點需求：

• 易用，Flink 提供了豐富的數(shù)據結構、多樣的狀態(tài)組織形式以及簡潔的擴展接口，讓狀態(tài)管理更加易用；
• 高效，實時作業(yè)一般需要更低的延遲，一旦出現(xiàn)故障，恢復速度也需要更快；當處理能力不夠時，可以橫向擴展，同時在處理備份時，不影響作業(yè)本身處理性能；
• 可靠，Flink 提供了狀態(tài)持久化，包括不丟不重的語義以及具備自動的容錯能力，比如 HA，當節(jié)點掛掉后會自動拉起，不需要人工介入。

4.狀態(tài)后端

做狀態(tài)數(shù)據（持久化，restore）的工具就叫做狀態(tài)后端。比如在 Flink 中見到的 RocksDB、FileSystem 的概念就是指狀態(tài)后端。這些狀態(tài)后端就是實際存儲上面的狀態(tài)數(shù)據的。比如配置了 RocksDB 作為狀態(tài)后端，MapState 的數(shù)據就會存儲在 RocksDB 中。

總的來說可以這么理解：應用中有一份狀態(tài)數(shù)據，把這份狀態(tài)數(shù)據存儲到 MySQL 中，這個 MySQL 就能叫做狀態(tài)后端。

5.Checkpoint和Savepoint

概念：協(xié)調整個任務 when，how 去將 Flink 任務本地機器中存儲在狀態(tài)后端的狀態(tài)去同步到遠程文件存儲系統(tǒng)（比如 HDFS）的過程就叫 Checkpoint、Savepoint。

Flink 狀態(tài)保存主要依靠 Checkpoint 機制，Checkpoint 會定時制作分布式快照，對程序中的狀態(tài)進行備份。分布式快照 Checkpoint 完成后，當作業(yè)發(fā)生故障了如何去恢復？假如作業(yè)分布跑在 3 臺機器上，其中一臺掛了。這個時候需要把進程或者線程移到 active 的 2 臺機器上，此時還需要將整個作業(yè)的所有 Task 都回滾到最后一次成功 Checkpoint 中的狀態(tài)，然后從該點開始繼續(xù)處理。

Checkpoint流程如下：

1. JM 定時調度 Checkpoint 的觸發(fā)：JM CheckpointCoorinator 定時觸發(fā)，CheckpointCoordinator 會去通過 RPC 接口調用 Source 算子的 TM 的 StreamTask 告訴 TM 可以開始執(zhí)行 Checkpoint 了。
2. Source 算子：接受到 JM 做 Checkpoint 的請求后，開始做本地 Checkpoint，本地執(zhí)行完成之后，發(fā) barrier 給下游算子。barrier 發(fā)送策略是隨著 partition 策略走，將 barrier 發(fā)往連接到的所有下游算子（舉例：keyby 就是廣播，forward 就是直接送）。
3. 剩余的算子：接收到上游所有 barrier 之后進行觸發(fā) Checkpoint。當一個算子接收到上游一個 channel 的 barrier 之后，就停止處理這個 input channel 來的數(shù)據（本質上就是不會再去影響狀態(tài)了）

Savepoint 與 Checkpoint 類似，同樣是把狀態(tài)存儲到外部介質。當作業(yè)失敗時，可以從外部恢復。主要區(qū)別如下：