「Flink」理解流式處理重要概念

什麼是流式處理呢？

這個問題其實咱們大部分時候是沒有考慮過的，大多數，咱們是把流式處理和實時計算放在一塊兒來講的。咱們先來了解下，什麼是數據流。

數據流（事件流）

• 數據流是無邊界數據集的抽象
• 咱們以前接觸的數據處理，大多都都是有界的。例如：處理某天的數據、某個季度的數據等
• 無界意味着數據是無限地、持續增加的
• 數據流會隨着時間的推移，源源不斷地加入進來
• 數據流無處再也不
• 信息卡交易
• 電商購物
• 快遞
• 網絡交換機的流向數據
• 設備傳感器發出的數據
• …
• 這些數據都是無窮無盡的
• 每一件事情，均可以當作事件序列
• 數據流是有序的
• 數據的到來老是有個前後順序
• 數據流是不可變的
• 事件一旦發生，就不能被改變
• 它陳述了某一個時刻的事實
• 數據流是能夠重播的
• 爲了處理的一些問題、糾正過去的錯誤，能夠重跑數據流
• 藉助於Kafka，咱們能夠從新消費幾個月以前的原始數據流

流式處理

流式處理就是指實時地處理一個或多個事件流。它是一種編程範式。其餘編程領域，主要有3種編程範式：

1. 請求與響應
• 延遲最小的一種方式，響應時間要求亞毫秒級到毫秒之間
• 響應時間通常分穩定
• 發出請求，等待響應（大部分的JavaEE同窗，都是開發這一類編程範式的應用），其實就是OLTP
2. 批處理
• 特色：高延遲、高吞吐
• 通常是固定某個時刻開始啓動執行，讀取全部的數據，而後輸出接口
• 每次讀取到的都是舊數據
• 主要應用在DWH或BI中
3. 流式處理
• 特色：介於上述二者之間
• 流式處理可讓業務報告保持更新，持續響應

流的定義不依賴某個框架，只要儲蓄從一個無邊界數據集中讀取數據，並對它們進行處理生成結果，就是進行流式處理。重點是：整個過程必須是持續的。

流式處理中的時間

上述咱們已經說過了，數據流都是有序的。某一時刻的數據是肯定的。時間是流式處理中很是重要的概念。大部分流式應用的操做都是基於時間窗口的。

流式系統通常包含如下幾個時間概念（熟悉Flink的同窗應該會很熟悉）：

• 事件時間（Eventtime）
• 事件實際發生的時間
• 用戶通常只對事件發生時間感興趣
• 日誌追加時間
• 日誌追加時間是指事件保存到事件存儲源的時間
• 例如：數據是什麼到達Kafka的（Kafka是能夠啓用自動添加時間戳功能的）
• 處理時間
• 流式處理應用接收到事件後，要對齊進行處理的時間
• 處理時間取決於流式處理應用什麼時候讀取到這個時間
• 若是應用程序使用了兩個線程來讀取同一個事件，這個時間戳可能會不同
• 這個時間戳很是不可靠，應該避免使用它

狀態

若是流式處理是來一個事件就處理一個事件，那麼流式處理就很簡單。但若是操做中包含了多個事件，流式處理就有意思了。例如：咱們想在流式處理中統計北京用戶的訂單數量、消費金額等等。此時，就不能光處理單個事件了，咱們須要獲取更多的事件。事件與事件之間的信息就稱之爲狀態。例如簡單的，求某個類型的訂單數等。

這些狀態通常就保存在流式處理程序本地變量（本地內存）中，例如：使用HashMap來保存計數。但這種作法是很不可靠的，流式處理處理的是無界數據集，一旦應用程序出現異常，就會出現狀態丟失，這是咱們說不能接受的。因此，每一種流式計算框架都會很當心地持久化狀態。若是應用程序重啓，須要將這些數據恢復。

流式處理通常包含兩種狀態：

• 本地狀態
• 這種狀態只能被應用程序實例訪問（不過Flink 1.9版本是能夠外部來訪問本地狀態的）
• 內嵌到應用程序的數據庫中進行維護和管理
• 特色：速度快，但受內存大小的限制，因此，不少流式處理系統都將數據拆分到多個子流中處理
• 外部狀態
• 用外部存儲來處理，通常使用NoSQL系統，例如：Cassadra
• 特色：沒有大小限制，能夠被應用程序多個實例訪問、甚至外部應用訪問，但引入額外的系統會形成延遲、複雜性（例如：要維護內部和外部狀態一致性問題）

時間窗口

大部分針對流的操做都是基於時間窗口的。例如：計算一週內銷量最好的產品。兩個流的合併也是基於時間窗口的。流式系統會合併發生在相同時間段上的事件。窗口是有類型的。如下幾點是咱們設計窗口須要考慮的：

• 窗口的大小
• 是基於5分鐘計算仍是基於15分鐘、甚至是一天
• 窗口越小，就能越快地發現變動，不過噪聲也就越多
• 窗口越大，變動就跟平滑，不過延遲也越嚴重
• 窗口的移動頻率（移動間隔）
• 5分鐘的窗口，能夠1分鐘計算一次，或者每秒鐘計算一次，或者每當有新事件到達時計算一次
• 若是「移動頻率」與窗口大小相等，這種稱爲滾動窗口（tumbling window）
• 若是窗口隨着每一條記錄移動，這種狀況稱爲滑動窗口（sliding window）
• 窗口的可更新時長
• 假設：計算了 00:00 – 00:05 之間的訂單總數，一個小時後，又獲得了一些「事件時間」是 00:02的事件（例如：由於網絡通訊故障，這個消息晚到了一段時間），這種狀況，是否須要更新 00:00 – 00:05 這個窗口的結果呢？或者就不處理了？
• 理想狀況下，能夠定義一個時間段，只要在這個時間段內，事件能夠被添加到對應的時間片斷裏。例如：若是事件處於4個小時之內，就更新，不然，就忽略掉。
• 窗口時間對齊
• 窗口能夠與時間對齊，例如：5分鐘的窗口若是每分鐘移動一次，那麼第一個分片能夠是：00:00 – 00:05，第二個就是 00:01 – 00:06
• 窗口也能夠不與時間對齊，例如：應用能夠在任什麼時候間啓動，那麼第一個分片有多是03:17 – 03:22
• 滑動窗口永遠不會與時間對齊，只要有新的記錄到達，就會發生移動

下面這張圖，說明了滾動窗口與滑動窗口的區別。

滾動窗口：假設窗口的大小爲5分鐘，這裏肯定的3個時間窗口

滑動窗口：假設每分鐘滑動一次，那麼這個時候會有5個時間窗口，計算結果會發生重疊

流式處理的設計模式

單個事件處理

這是流式處理最基本的模式。這種模式也叫：map或filter模式。常常被用來過濾無用的事件或者用於轉換事件。

這種模式，應用程序讀取流中的數據，修改數據，而後把事件生成到另外一個流上。這一類應用程序無需在程序內部維護狀態，每個事件都是獨立處理的。這種錯誤恢復和進行負載均衡都很容易。由於無需進行狀態恢復操做。

使用本地狀態

大部分流式處理應用關係如何聚合數據。特別是：基於時間窗口進行聚合。例如：找到天天最低、最高的交易價格。要實現這種操做，就須要維護流的狀態。例如：咱們須要將最小值、最大值保存下來，用它們與每個新值對比。這類操做，能夠經過本地狀態來實現。例如：每個分組都維護本身分組的狀態。

一旦流式處理中包含了本地狀態，就須要解決如下問題。

• 內存使用
• 必需要有足夠的內存來保存本地狀態
• 持久化
• 確保應用程序關閉時，不會丟失狀態
• 例如：咱們可使用RocksDB將本地狀態保存到內存裏、同時持久化到磁盤上，以便重啓後恢復。並且須要將本地狀態的變動發送到Kafka的主題上
• 從新負載均衡
• 有時候，分區被從新分配給不一樣的消費者。這種狀況，失去分區的實例必須把最後的狀態保存下來，或得分區的實例必需要知道如何恢復到正確的狀態

多階段處理和重分區

有些時候，咱們要經過全部可用的數據來得到結果。例如：要發佈天天的「前10支」股票，這10支股票須要從天天的交易股票中挑選出來。若是僅僅在單個實例上處理是不夠的，由於10支股票分佈在多個實例上。

此種，咱們分爲多個階段來處理。

一、計算每支股票當天的漲跌。這個計算能夠在每一個實例上執行

二、將結果寫入到單個分區

三、再用一個實例找出當天的前10支股票

這一類操做就與MapReduce很像了。

使用外部查找——流和表的鏈接

有時候，流式處理須要將外部數據和流集成在一日。例如：外部數據中保存了一些規則、或者將完整完整地用戶信息拉取到流中。

這種case最大的問題，外部查找會帶來嚴重的延遲，通常在 5-15 ms之間，這在不少狀況下是不可行的。並且，外部系統也沒法承受這種額外的負載——流式處理系統每秒能夠處理10-50W個事件，而數據庫正常狀況下每秒只能處理1W個事件，因此須要伸縮性更強的解決方案。

爲了獲取更好的性能和更強的伸縮性，須要將外部數據庫的信息緩存到流式處理應用中。但考慮如下問題：

如何保證緩存裏的數據是最新的？

若是刷新太頻繁，仍然會對數據庫形成很大壓力，緩存也就無用了。

若是刷新不及時，那麼流式處理中所用的數據就會過期。

若是可以捕捉數據庫的變動事件，並造成事件流，流式處理做業就能夠監聽事件流，並及時更新緩存。捕捉數據庫的變動事件並造成數據流，這個過程稱爲CDC（Change Data Capture）。例如：咱們能夠經過Canal來捕獲MySQL數據庫的變化、能夠經過ogg來捕獲Oracle數據庫的變化

流與流的鏈接

有時候須要鏈接兩個真實的事件流。要鏈接兩個流，就是鏈接全部的歷史事件（將兩個妞中具備相同鍵、發生在相同時間窗口內的事件匹配起來），這種流和流的鏈接稱爲：基於時間窗口的鏈接（windowed-join）。鏈接兩個流，一般包含一個滑動時間窗口。

亂序事件

無論對於流式處理、仍是傳統的ETL系統，處理亂序事件都是一個挑戰。物聯網領域常常發生亂序事件：一個移動設備斷開Wifi鏈接幾個小時，在從新連上WiFi後，將幾個小時堆積的事件一併發出去。要讓流式處理應用處理好這些場景，須要作到幾下：

• 識別亂序事件
• 應用程序須要檢查事件的時間，並將其與當前時間進行比較
• 規定一個時間段用於重排亂序事件
• 例如：3個小時之內的事件能夠重排，但3個小時之外的事件就能夠直接扔掉
• 具備必定時間段內重排事件的能力
• 這是流式處理應用和批處理的重要不一樣點
• 假設有一個天天運行的做業，一些事件在做業結束以後纔到達，那麼能夠從新運行昨天的做業來更新
• 而在流式處理中，從新運行昨天的做業是不存在的，亂序事件和新到達的事件必須一塊兒處理
• 具有更新結果的能力
• 若是處理的結果保存在數據庫你，那麼能夠經過put或update對結果進行更新

從新處理

該重要模式是從新處理事件：

• 流式處理應用更新了，要使用新版本應用處理同一個事件流，生成新的結果，並比較兩種版本的結果，而後某個時間點將客戶端切換到新的結果流
• 現有的流式處理出現了缺陷，修復後，須要從新處理並從新計算結果

第一種狀況，須要Kafka將事件流長時間地保存在可伸縮的數據存儲中

• 將新版本的應用做爲一個新的消費者組
• 新的版本從輸入主題的第一個偏移量開始讀取數據
• 檢查結果流，在新版本的處理做業遇上進度時，將客戶端應用程序切換到新的結果流上

第二種狀況，須要應用程序回到輸入流的起始位置開始處理，同時重置本地狀態，還要清理以前的輸出流。這種方式處理起來比較困難。建議仍是使用第一種方案。

參考文獻：

《Kafka全文指南》