淺談一下流式處理平臺（Flink）

大資料框架聽過很多，比如 Hadoop，HDFS...不過自己的專案都沒有上過

為什麼突然提到 Flink，因為最近一個專案需要用到，所以學習最好的方式就是專案驅動

以前總覺得自己要學會了某樣東西再去做，等學會了，也許又用不上，久而久之，又忘了

下面我結合專案，淺談一下 Flink

01、Flink 入門

百度一下：Flink 是什麼？

上面的介紹我們每個字都看得懂，但連在一起就看不懂了。

不管怎麼樣，至少我們知道了：Flink 是一個分散式的計算處理引擎。

• 分散式：它的儲存或計算交由多臺伺服器完成，最後彙總起來達到最終的效果。
• 實時：處理速度是毫秒或者秒級的。
• 計算：可以簡單理解為是對資料進行處理。

什麼是有邊界和無邊界？

Apache Flink 是一個框架和分散式處理引擎，用於在*無邊界和有邊界*資料流上進行有狀態的計算。

官網其實也有介紹，但對初學者來說不太好理解，我來幼兒園化一下。

大家應該都用過訊息佇列吧，Producer 生產資料，發給 Broker，Consumer 消費，完事。

在消費的時候，我們需要知道 Producer 發了多少條訊息嗎？什麼時候發的嗎？不需要吧。反正你來一條，我就處理一條，沒毛病吧。

這種沒有做任何處理的訊息，預設就是無邊界的。

那有邊界就很好理解了嘛：無邊界的基礎上加上條件，那就是有邊界的。

加什麼條件呢？比如我要加個時間：我要從3月22號

什麼時候用有邊界，什麼時候用無邊界？那也很好理解，

我做資料清洗：來一條，我處理一條，這種無邊界就好了。

我要做資料統計：每小時的pv(page view)是多少，那我就設定1小時的邊界，攢夠1小時我再來處理。

在Flink上，設定“邊界”這種操作叫做開視窗（window），視窗可以簡單分為兩種型別：

• 時間視窗（TimeWindow）：按照時間視窗對資料進行聚合，比如上面說的1小時一次。

• 計數視窗（CountWindow）：按照指定的條數來做聚合，比如攢夠10條資料處理一次。

聽著就很人性化（媽媽再也不用擔心我需要聚合了）

不僅如此，在Flink

比如：我在 11:06分 產生了5條資料，在11:07產生了4條資料，我現在是按時間視窗每分鐘來做聚合的。

理論上來講：Flink應該在 11:06分聚合了5條資料，在11:07聚合了4條資料。

但是，可能由於網路的延遲性原因，導致11:06分有3條資料在11:07才接收到，如果不做處理，就造成了 06分處理3條，07分處理7條。

這個結果對於需要準確性的場景來說，就太不合理了。所以Flink可以給我們指定時間語義，不指定預設是資料到Flink的時候（Process Time）來進行處理。我們可以指定聚合時間為事件發生的時間（Event Time）來進行處理。

事件發生時間就是：日誌真正記錄的時間。

2020-11-22 00:00:02.552 INFO  [http-nio-7001-exec-28] c.m.t.rye.admin.web.aop.LogAspect

雖然指定了聚合時間為事件發生的時間（Event Time），但還是沒解決資料亂序的問題（06分產生了5條資料，實際上06分只收到了3條，而剩下的兩條在07分才收到，那此時怎麼辦呢？在06分時該不該聚合，07分收到的兩條06分資料怎麼辦？）

Flink又給我們設定了水位線（waterMarks）

Flink的意思就是：存在網路延遲等情況導致資料接受不是有序，這種情況我能理解，那你設定一個延遲時間吧，等延遲時間到了，我再做統一聚合。

比如我設定延遲時間為1分鐘，等到07分的時候，我再對 06分的視窗做聚合處理。

什麼叫有狀態？

Apache Flink 是一個框架和分散式處理引擎，用於在無邊界和有邊界資料流上進行有狀態的計算。

無狀態我們可以理解為：每次的執行都不會依賴以前的執行結果，每次的執行都是獨立的。

有狀態就很好理解了：每次的執行都依賴上一次的執行結果。

舉個栗子：我們現在要統計文章的閱讀pv(page view)，現在只要有一個點選了文章，在kakfa中就會有一條訊息。現在我要在流式處理平臺上進行統計，那此時是有狀態還是無狀態的？

比如說：我用Redis對結果進行儲存，來一條資料，我就先查一下Redis目前的值是多少，然後在當前值的基礎上累加一下就完事了。

假設用Flink做，Flink本身就提供了這種功能給我們使用，我們可以依賴Flink的“儲存”，將每次的處理結果交給Flink處理。

可以簡單的認為：Flink本身就給我們提供了“儲存”的功能，我們的每次執行是可以依賴Flink的“儲存”的，所以它是有狀態的。

那Flink是把這些有狀態的資料儲存在哪的呢？

主要有三個地方：

• 記憶體
• 檔案系統（HDFS）
• 本地資料庫

如果Flink掛了，可能記憶體的資料沒了，磁碟可能儲存了部分的資料，那再重啟的時候，就不怕資料丟失嗎？

看到這裡，你可能會在別的地方看過Flink的另外一個比較出名的特性：精確一次性（exactly once）

什麼是精確一次性（exactly once）？

眾所周知，流的語義性有三種：

• 精確一次性（exactly once）：有且只有一條，不多不少
• 至少一次（at least once）：至少有一條，只多不少
• 最多一次（at most once）：最多有一條，可能沒有

Flink的精確一次性指的是：狀態（這個狀態指上面提到的有狀態）只持久化一次到最終介質中（本地資料庫/HDFS...）

以上圖為例：假設我的 source 有以下數字21,13,8,5,3,2,1,1，然後在Flink中做一個累加操作。

現在已經處理完2,1,1了，所以累加的 sum = 4，現在Flink把狀態 sum 已經儲存到最終介質中去了

然後開始處理5,3，得到的 sum = 12，但是 Flink還沒來得及把 12儲存到最終介質中 就掛了。

Flink重啟後會重新把系統恢復到 sum = 4 的狀態，所以5,3得繼續計算一遍。

所以 Flink的精確一次性指的就是：狀態只持久化一次到最終介質中（本地資料庫/HDFS...）

至於Flink是在多長時間儲存一次，是我們自己手動配置的。

根據上面的程式碼，我們可以發現，設定儲存的時間在Flink中叫做 CheckPoint。

所謂的 CheckPoint就是 Flink會在指定的時間段上儲存狀態的資訊，如果Flink掛了，可以直接將上一次儲存的狀態資訊撈出來，繼續執行計算，最終實現（exactly once）。

那CheckPoint是怎麼實現的呢？

想想我們使用 kafka 在業務上的至少一次是怎麼實現的：我們從 kafka 上面拉取資料，處理完業務後，手動提交 offset（告訴kafka 我已經處理完了）

我們是做完了業務後才將offset進行 commit 的， checkpoint其實也是一樣的（等拉下來的資料全部處理完，才進行真正的 checkpoint）

那Flink是怎麼知道拉下來的資料已經走完了呢？

Flink在流處理過程中插入了 barrier，每個環節處理到 barrier 都會上報給 sink，等所有的 barrier 都上報了，說明這次的 checkpoint已經走完了

02、什麼是流式處理平臺

舉個例子，比如商家要在平臺上投放廣告，我們要給商家看到廣告帶來的效果，最核心的就是 「曝光量」「點選量」「訂單量」

下面來聊一下這個「發展歷程」，看完這個過程或許可以更好的瞭解為什麼需要流式處理平臺

1、PHP階段：在最初時，業務以及系統結構都比較簡單，直接把「曝光量」「點選量」「訂單量」存入資料庫，再一把梭通過定時任務進行全量聚合，得到最終的效果資料。

在這個階段裡，由於資料量不大，通過定時任務全量來做聚合資料也不算不可以，那時候商家都能接受業務的延遲性，大概5min出資料。

2、Java階段：隨著業務發展，資料量日益提升，站內中介軟體服務平臺也發展起來了。通過中介軟體團隊提供的消費 binlog框架，從架構上改變聚合模式。這個階段可以更快的給商家展示效果資料，大概1min出資料。

3、流式處理平臺階段：流式處理平臺是對「計算」或者說是處理資料時的抽象，在這抽象基礎上，它更能充分利用系統的資源（一個大的任務拆分成多個小任務，分發到不同的機器上執行），大概秒級出資料。

02、專案哪些地方用到了 Flink

我所做的專案是一個訊息傳送系統，Flink 用於對「訊息傳送者」「模板訊息」等維度作聚合處理，得到的資料再給到介面去展示或者排查問題使用，能大大提高排查方和業務方的使用效率

最後，給大家安利一個公眾號：Java3y，備註【專案】，進群一起跟著 3y 做專案，專案就是當前的訊息傳送系統，公眾號上還有很多面試資料，快衝啊。