從盤古開天辟地說起為什么 Flink CP 能實現(xiàn)精確一次之二
為什么我要把這段話放在最前面呢?因為博主有了大發(fā)現(xiàn)，博主在總結學習的過程中，總結了除了 Flink CP、Chandy-Lamport 全局一致性快照算法之外的一種 通用全局一致性快照算法!!!。
這套 通用算法 包含 Chandy-Lamport 算法 ≈ Flink 非對齊 CP 算法 包含 Flink 對齊 CP 算法。
可能這一套 通用算法 之前已經(jīng)有人提過了，但是博主是自己在總結 Flink CP、Chandy-Lamport 算法的過程中，逆推總結出來的，并沒有借助外力!!!
1.前言
對于很多做離線或者實時數(shù)倉的小伙伴來說，我先問幾個問題，看看小伙伴萌能回答上來嗎?
- ? 你知道狀態(tài)是什么嗎?在離線數(shù)據(jù)開發(fā)的經(jīng)歷中，你碰到過狀態(tài)的概念嗎?
- ? 為什么離線數(shù)倉不需要狀態(tài)，實時數(shù)據(jù)開發(fā)中老是提到狀態(tài)的概念?
- ? Flink 中的狀態(tài)、狀態(tài)后端、全局一致性快照(Checkpoint\Savepoint) 的作用都是什么，這三個概念的關聯(lián)又是什么?
- ? Flink 是通過什么機制來做 Checkpoint 的?為什么這套機制能夠做到故障恢復呢?
- ? Flink Checkpoint 是基于 Chandy-Lamport 算法的，但是 Flink 的實現(xiàn)相比 Chandy-Lamport 算法之間又有哪些優(yōu)點、缺點?
- ? Flink Checkpoint 用到了 barrier，為什么用了 barrier 做的快照就能保證全局一致性快照的正確性?barrier 到底起到了什么作用?
- ? Flink 對齊 Checkpoint 和非對齊 Checkpoint 的區(qū)別是什么?非對齊 Checkpoint 也能保障精確一次嗎?
小伙伴們思考一下，都能回答上來么，如果對于某些問題你還有疑問，樓主會通過本篇文章幫你解答這些問題，理清這些概念!
由于本文內(nèi)容較多，所以博主將本文分為上，中，下三集，本文是中，三集內(nèi)容是有連接關系的，如果小伙伴在看本文的過程中對有些概念不清楚，可以跳轉(zhuǎn)到上文進行查看：
本文最主要的內(nèi)容就是解釋了：
一個分布式應用是怎么異步做一個全局一致性快照?
2.名詞解釋
- ? Single-Token conservation：一個最常見的分布式應用，單 Token 流轉(zhuǎn)分布式應用
- ? Process：指分布式應用中的進程，舉個 Flink 中的例子就是 TaskManager
- ? Channel：指分布式應用中進程之間的傳輸通道，舉個 Flink 中的例子就是 TaskManager 之間傳輸數(shù)據(jù)的網(wǎng)絡傳輸通道
3.分布式應用全局一致性快照要記錄的狀態(tài)內(nèi)容
首先在分析一個復雜的大數(shù)據(jù)應用的全局一致性快照之前，我們先以最簡單的分布式應用為例。
Single-Token conservation：其有 p 和 q 兩個進程，p 可以通過 Channel pq(記為 Cpq) 向 q 發(fā)消息，q 可以通過 Channel qp(記為 Cqp) 向 p 發(fā)消息，其中有一個叫 token 的消息，在這個系統(tǒng)中一直不停的傳輸流轉(zhuǎn)，從 p 到 q，再從 q 到 p。
- ? 首先我們來分析這個應用中，全局一致性快照應該包含哪些內(nèi)容?
- ? 結論：全局一致性快照 = Process 狀態(tài) + Channel 狀態(tài)。
- ? 原因：以上面的四幅圖為例，每一幅圖代表一個時刻，如果我們以拍照這種方式做全局一致性快照來理解時，那么同一時刻，Process 和 Channel 同時都會存在數(shù)據(jù)，這些數(shù)據(jù)都是作為全局一致性快照的一部分內(nèi)容。
使用上述的這個結論，我們可以得到上圖 Single-Token conservation 示例中的全局一致性快照 S = S(p) + S(Cpq) + S(q) + S(Cqp)
其中：
- S：全局一致性快照
- S(p)：p 進程的狀態(tài)
- S(Cpq)：p 進程到 q 進程的 Channel 狀態(tài)
- S(q)：q 進程的狀態(tài)
- S(Cqp)：q 進程到 p 進程的 Channel 狀態(tài)
這里就碰到了我們要分析的關鍵問題：做全局一致性快照時，小伙伴萌都容易理解S(p)，S(q)這兩個，因為這兩份狀態(tài)數(shù)據(jù)就真實的存在我們的分布式應用中，但是S(Cpq)，S(Cqp)這兩個怎么理解呢?這些數(shù)據(jù)都是在網(wǎng)絡中傳輸啊，我們做全局一致性快照時用啥方法才能把這些數(shù)據(jù)也記錄下來呢?接下來詳細講講博主的理解
4.Process 狀態(tài)記錄的內(nèi)容
記錄和實際業(yè)務相關的狀態(tài)內(nèi)容。舉例：id 去重就存儲歷史所有的 id 就可以了。
5.Channel 狀態(tài)記錄的內(nèi)容
還是以前文的 Single-Token conservation 為例：
token 在 p 時(對應第一張圖)，這時的全局一致性快照為：
S(token-in-p) = S(p-token-in-p) + S(Cpq-token-in-p) + S(q-token-in-p) + S(Cqp-token-in-p)
其中：
- S(token-in-p)：token 在 p 時，做的全局一致性快照
- S(p-token-in-p)：token 在 p 時，p 進程的狀態(tài)
- S(Cpq-token-in-p)：token 在 p 時，p 進程到 q 進程的 Channel 狀態(tài)
- S(q-token-in-p)：token 在 p 時，q 進程的狀態(tài)
- S(Cqp-token-in-p)：token 在 p 時，q 進程到 p 進程的 Channel 狀態(tài)
其中 S(p-token-in-p) 好理解，做快照時，token 還沒有從 p 發(fā)出去，p 肯定知道 token 還在 p;但是站在 Cpq 做狀態(tài)時來說：Cpq 做狀態(tài)時，怎么保障 Cpq 知道 token in p?
在分析上面這個問題前，博主先使用 數(shù)學的方式 分析一下 S(Cpq) 到底記錄了哪些內(nèi)容。
? 第一步：定義變量
- n：在 p 的狀態(tài)記錄前，p 記錄的 p 發(fā)往 Cpq 的 msg 數(shù);
- n′：在 Cpq 的狀態(tài)記錄前，Cpq 記錄的 p 發(fā)往 Cpq 的 msg 數(shù);
- m：在 q 的狀態(tài)記錄前，q 記錄的 q 從 Cpq 中接收到的 msg 數(shù);
- m′：在 Cpq 的狀態(tài)記錄前，Cpq 記錄的 q 從 Cpq 中接收到的 msg 數(shù);
? 第二步：提出假設
- 假設 Channel 和 Process 一樣，也可以自主的去將做快照時 Channel 中進行網(wǎng)絡傳輸?shù)臄?shù)據(jù)作為狀態(tài)保存下來;對應到上述案例中就是 Cpq 可以主動的將做好的 S(Cpq) 狀態(tài)保存下來;
? 第三步：先說結論
- Cpq 記錄S(Cpq)時，必然會有 n = n' ≥ m = m';
- 一個 Channel 要記錄的狀態(tài)是，它 sender 記錄自己狀態(tài)之前 channel 所接收到 sender 發(fā)的的 msg 列表，再減去 receiver 記錄自己狀態(tài)之前 channel 已經(jīng)發(fā)給 receiver 的 msg 列表，減去的之后的 msg 就是還在 Channel 中的數(shù)據(jù)，這些數(shù)據(jù)是需要 Channel 作為狀態(tài)記錄下來的。
- 而如果 n′ = m′，那么 Channel c 中要記錄的 msg 列表就是 empty 列表。如果 n′ > m′，那么要記錄的列表是 (m′+1),…n′ 號消息對應的 msg 列表。
? 第四步：給出證明
首先：n = n'，利用反證法：如果 n != n'，則會有：
1.n > n' 時，假設：
- n = 10(p 記錄狀態(tài)前，p 記錄 p 發(fā)往 Cpq msg 數(shù)為 10(msg 編號 1 - 10));
- n' = 7(Cpq 記錄狀態(tài)前，Cpq 記錄 p 發(fā)往 Cpq 的 msg 數(shù)為 7(msg 編號 1 - 7));
- 那么假設 token 這條 msg 的編號為 9，就會出現(xiàn) p 記錄的狀態(tài)為S(p-token-in-Cpq)，Cpq 記錄的狀態(tài)為S(p-token-in-p)，實際這是不符合全局一致性快照的要求的;
2.n < n' 時，假設：
- n = 7(p 記錄狀態(tài)前，p 記錄 p 發(fā)往 Cpq msg 數(shù)為 7(編號 1 - 7));
- n' = 10(Cpq 記錄狀態(tài)前，Cpq 記錄 p 發(fā)往 Cpq 的 msg 數(shù)為 10(編號 1 - 10));
- 那么假設 token 這條 msg 的編號為 9，就會出現(xiàn) p 記錄的狀態(tài)為S(p-token-in-p)，Cpq 記錄的狀態(tài)為S(p-token-in-Cpq)，實際這是不符合全局一致性快照的要求的;
3.n = n' 時，假設：
- p 做出S(p-token-in-p)的狀態(tài)時，因為 n = n'，這就代表 p 沒有把 token 發(fā)出去，Cpq 也沒有接受到 token，Cpq 就知道 token 沒有發(fā)過來，則只有這種情況可以滿足S(Cpq-token-in-p);
其次：m = m'，同樣利用反證法可以得到，下文只舉 m > m' 的案例：
1.m > m' 時：
- n = n' = m = 10(q 記錄狀態(tài)前，Cpq 記錄 q 從 Cpq 接收到的 msg 數(shù)為 10(編號 1 - 10));
- m' = 7(Cpq記錄狀態(tài)前，Cpq 記錄的 q 從 Cpq 接收到的 msg 數(shù)為 7(編號 1 - 7));
- 那么假設 token 這條 msg 的編號為 9，就會出現(xiàn) Cpq 記錄的狀態(tài)為S(Cpq-token-in-Cpq)，q 記錄的狀態(tài)為S(q-token-in-p)，實際這是不符合全局一致性快照的要求的;
最后：n' ≥ m' and n ≥ m：同樣可以利用反證法得到，此處不再舉例。
4.? 第五步：解答 2.4 節(jié)提出的 Cpq 怎么知道 token in p 的問題
通過 n = n' ≥ m = m' 其實就可以推論出 Cpq 一定會知道 token in p。
為了幫大家更容易的理解一個分布式應用包含的全局一致性快照包含的數(shù)據(jù)內(nèi)容，接下來我用偽代碼描述一下，會比文字更好理解~
6.偽代碼描述一個分布式應用全局一致快照包含的數(shù)據(jù)內(nèi)容
偽代碼如下：
// S_all 即一個分布式應用的全局一致性快照
S_all = null;
// 假設總共有 x 個 Process，S_all 先把所有 Process 的狀態(tài)記錄下來
for (int i = 1; i <= x; i++) {
// 第 i 個 Process 的狀態(tài)為 S_P_i，直接按照 += 寫，勿噴
S_all += S_P_i;
}
// 假設總共有 y 個 Channel，S_all 把所有 Channel 的狀態(tài)記錄下來
for (int i = 1; i <= y; i++) {
// 1. S_C_i：第 i 個 Channel 的狀態(tài)
// 2. m_i：第 i 個 Channel 做快照前，發(fā)往下游 Process 的消息（數(shù)據(jù)）編號，m_i 其實就是上文變量中的 m
// 3. n_i：第 i 個 Channel 做快照前，接受上游 Process 的消息（數(shù)據(jù)）編號，n_i 其實就是上文變量中的 n
// 4. 需要注意，每一個 Channel 的 m_i 和 n_i 的數(shù)值都可能是不一樣的
// 5. Message[m_i + 1] ：代表編號為 m_i + 1 的那條消息（數(shù)據(jù)）。舉例 Message[0] 代表編號為 0 的那條消息（數(shù)據(jù)）
S_C_i = Message[m_i + 1] + ... + Message[n_i];
S_all += S_C_i;
}
// 狀態(tài)做完了
7.怎樣去記錄 Channel 的狀態(tài)?
通過上面的分析，我們已經(jīng)討論得到了S(Cpq)都包含了什么內(nèi)容，并且其之間要滿足什么樣的數(shù)學關系。但是在現(xiàn)實實際生活中，消息在 Channel 上傳輸(光纖上傳輸)時，我們是無法記錄這些消息作為 Channel 的狀態(tài)的。
那么有沒有什么思路可以讓我們也能夠去記錄 Channel 的消息呢?
當然有。
因為只要我們分布式應用的傳輸這些消息的光纖沒有被挖斷，消息終究會通過 Channel 到達 Process 的，因此我們就可以自然的想到其實可以在消息傳輸 終點的 Process 去記錄這些消息作為 Channel 的狀態(tài)。對應到上述的 Single-Token conservation 案例來說，我們可以在 q 中記錄 Channel pq 的S(Cpq)，在 p 中記錄 Channel pq 的S(Cqp)。
如果是按照這個思路去分析的話，上面那段偽代碼就可以簡化為下面這樣：
// S_all 即全局一致性快照
S_all = null;
// 假設總共有 x 個 Process
for (int i = 1; i <= x; i++) {
// S_i_all：第 i 個 Process 要記錄的所有狀態(tài)
S_i_all = null;
// S_P_i：第 i 個 process 的狀態(tài)
S_i_all += S_P_i; // 【直接按照 += 寫，勿噴】
// 第 i 個 Process 總共有 y 個輸入 channel，下文中 j 即指代第 i 個 Process 的第 j 個上游 Channel
for (int j = 1; j <= y; j++) {
// 1.S_C_j：第 j 個 Channel 的狀態(tài)
// 2.m_j：第 j 個 Channel 做快照前，發(fā)往下游 Process 的消息（數(shù)據(jù)）編號
// 3.n_j：第 j 個 Channel 做快照前，接受上游 Process 的消息（數(shù)據(jù)）編號
S_C_j = Message[m_j + 1] + ... + Message[n_j];
S_i_all += S_C_j;
}
S_all += S_i_all;
}
// 狀態(tài)做完了
解釋一下上面的偽代碼：
1.S_all：所有的進程記錄的狀態(tài)之和，即所有S_i_all之和
2.S_i_all：每一個進程要記錄的所有狀態(tài)之和
3.Process i 在做S_i_all其實只有一個變量在做快照時是不知道到的，那就是 n_j(即第 i 個 channel 做快照前，接受到 j(上游) 的消息個數(shù))
- S_P_i是 Process 自己的狀態(tài)，所以是明確已知的
- S_C_j是 Channel 狀態(tài)，由Message[m_j + 1] + ... + Message[n_j]組成，其中 m_j 的含義是這個 Channel j 在做快照時發(fā)給當前 Process i 的消息編號，前文已經(jīng)介紹到 m = m'，即 m_j 值就等于是是當前 Process i 在做S_P_i時，接收到上游 Channel 的消息編號，所以 m_j 是明確已知的。但是 Process i 在做上游 Channel j 快照時 n_j 是無法獲取到的。
那么 Process i 在做S_C_j如何獲取到 n_j 呢?
這里我們就可以想到 n = n'：
- 因為 n = n'，所以 n_j 的值就等于 Channel j 的上游 Process 在做快照時，Process 發(fā)往 Channel j 的消息個數(shù);
- 所以我們可以認為 n_j 是 Channel j 的輸入 Process 在做快照時已知的一個值，則只需要這個 Process 把這個值發(fā)給 Process i 就行;
- Channel j 的輸入 Process 告知 Process i 的方式其做完快照之后，直接發(fā)一個 marker 下去，這個 marker 不會對計算有任何影響(即不會對狀態(tài)產(chǎn)生任何影響)，marker 只是一個標識，當 Process i 接受到這個 marker 時，就知道 n_j 的具體值了。
大家注意到了沒，這個 marker 其實就對應到了 Flink 中 Checkpoint 的 barrier。
8.總結
本文主要講了一個分布式應用異步做一個全局一致性快照的機制。
好，今天主要就講這么多，下集我們再說說 Chandy-Lambort，F(xiàn)link CP 和今天介紹的全局一致性快照原理的關系。
