Flink流式處理概念簡介

一，抽象層次

Flink提供不同級別的抽象來開發流/批處理應用程序。

1，stateful streaming

最底層。它通過Process Function嵌入到DataStream API中。它允許用戶從一個或多個流自由處理事件，並使用一致的容錯狀態。此外，用戶可以註冊事件時間和處理時間回調，允許程序實現複雜的計算。

2，Core APIs

實際上，大多數應用程序不需要上述的低級別抽象，而是針對Core API（如DataStream API（有界/無界流））和DataSet API（有界數據集）進行編程。這些流暢的API為數據處理提供了常見的構建模塊，如用戶指定的各種轉換形式，連接，聚合，窗口，狀態等。在這些API中處理的數據類型以各自的編程語言表示為classes。底層的Process Function和DataStream API的整合，使得針對一些特定的操作可以實現更低層次的抽象。DataSet API為有界數據集提供了額外的原函數，如循環/迭代。

3，Table API

Table API是以表為中心的聲明式DSL，可能是動態更改表（表示流時）。Table API遵循（擴展）關係模型：Table具有附加schema（與關係資料庫中的表相似），API提供操作，例如select，project，join，group-by，aggregate等。Table API代表的是應該做什麼邏輯操作，而不是直接指定如何編寫操作的源代碼。雖然Table API可以通過各種類型的用戶定義的函數進行擴展，但它不如Core API那麼具有表達力，但使用起來更加簡潔（少寫很多代碼）。

此外，Table API程序還可以通過在執行之前應用優化規則的優化器。

可以在表和DataStream / DataSet之間無縫轉換，允許程序將Table API和DataStream和DataSet API混合使用。

4，SQL

最高層次的抽象就是SQL。無論是語法還是表達，該層次的抽象都很像Table API。SQL抽象與Table API緊密交互，SQL查詢可以在Table API中定義的表上執行。

二，Programs and Dataflows

Flink程序的基本構建塊是流和轉換。在概念上，stream是data records的（潛在的永無止境的）flow，並且變換是將一個或多個流作為輸入的操作，並且作為結果產生一個或多個輸出流。

執行時，Flink程序被映射成streaming dataflows，由streams和轉換操作符組成。每個dataflow從一個或多個sources開始，並以一個或多個sinks結束。dataflows像任意的有向無環圖(DAG)。雖然通過迭代構造允許特殊形式的循環，但是為了簡單起見，我們大部分都會任務是DAG。

通常，程序中的變換和數據流中的運算符之間存在一對一的對應關係。然而，有時，一個變換可能由多個轉換運算元組成。

三，Parallel Dataflows

Flink中的程序本質上是並行和分發的。在執行期間，流具有一個或多個流分區，並且每個運算符具有一個或多個運算符subtask。操作符subtask彼此獨立，並以不同的線程執行，可能在不同的機器或容器上執行。

運算符子任務的數量是該特定操作符的並行性。stream的並行性總是其生產運算符的並行性。同一程序的不同運算符可能具有不同的並行級別。

Streams可以以一對一（或轉發）模式或重新分配模式在兩個運算符之間傳輸數據：

1)，一對一

One-to-one streams（例如上圖中的Source和map（）運算符之間）保留元素的分區和ordering。這意味著map()運算符的子任務[1]看到的元素的順序將和由Source操作符產生的順序相同。

2)，再分配

重新分配流(如map()和上面的keyBy / window之間以及keyBy / window和Sink之間)改變流的分區。每個操作符子任務根據所選擇的轉換將數據發送到不同的目標子任務。

KeyBy是按照key的hash值進行重新分區，rebalance()是以隨機的方式重新分區。在重新分配交換中，元素之間的排序僅保留在每對發送和接收子任務對(例如，keyBy / window的map()和子任務[2]的子任務[1]中)。上例子中僅保證相同key的元素順序性。

四，Windows

Aggregating events（例如，counts，sums）在流上的工作方式與批處理不同。例如，不可能對流中的所有元素進行計數，因為流通常是無限的(無界)。相反，流上的聚合(計數，總和等)由窗口限定，例如「最後5分鐘計數」或「最後100個元素的總和」。

Windows可以時間驅動（例如：每30秒）或數據驅動（例如：每100個元素）。通常區分不同類型的窗口，例如翻轉窗口（無重疊）[tumbling windows (no overlap)]，滑動窗口（具有重疊）[sliding windows (with overlap)]和會話窗口（由不活動空隙劃分）[ession windows (punctuated by a gap of inactivity)]。

五，Time

當在Streaming程序中使用時間的時候,如定義窗口時，可以參考不同的時間概念：

1，Event Time

Event Time是Event創建的時間。Events中的Event Time是以時間戳的格式存在。Fink通過timestamp assigners來獲得event timestamps。timestamp assigners後面出文章具體說。

2，Ingestion time

Ingestion time是event進入flink dataflow的時間。

3，Processing Time

Processing Time是基於時間的操作運算元執行的本地時間。

六，Stateful Operations

雖然dataflow中的許多操作只是一次處理一個單獨的事件（例如事件解析器），但一些操作會記住跨多個事件的信息（例如窗口運算符）。這些操作稱為有狀態。有狀態的操作運算元，狀態保存在嵌入式的鍵/值存儲中。狀態會和被狀態操作運算元讀取的streams一起分區和分配。使用keyBy函數後，僅keyed streams可能獲取key/value狀態，並且僅限於與當前事件key相關的值。對齊stream和state的keys，確保所有狀態更新都是本地操作，保證一致性，無需事務開銷。此對齊還允許Flink重新分配狀態並透明地調整流分區。

七，Checkpoints

Flink使用stream replay和checkpoint組合來實現容錯。檢查點與每個輸入流中的特定點相關聯，以及每個運算符的相應狀態。Streaming dataflow可以從檢查點恢複流，同時通過恢復操作符的狀態，從檢查點重新執行事件來保持一致性(一次性處理語義)。

檢查點間隔是在恢復時間（需要重新計算的事件數）的情況下，在執行期間消除容錯的開銷的一種手段。

八，Batch on Streaming

Flink執行批處理程序作為流程序的特殊情況，其中streams是有限的（有限數量的元素）。DataSet在內部被視為數據流。以上概念因此適用於批處理程序，同樣適用於流式傳輸程序，但有一些例外：

1，批處理程序的容錯不使用checkpoint。恢復需要完全執行流。這是可能的，因為輸入是有限的。恢復成本增加，但是使得處理變得廉價，因為無需進行checkpoint。

2，DataSet API中的有狀態操作使用簡化的in-memory/out-of-core數據結構，而不是鍵/值索引。

3，DataSet API引入了特殊的同步（superstep-based）迭代，這些迭代只能在有界流上進行。具體後面出文章介紹。

九，Tasks and Operator Chains

為了分散式執行，Flink鏈式合併operator subtasks成tasks。每個task被一個線程執行。鏈式的將操作符合併成任務是一個很有用的優化：它減少了線程到線程切換和緩衝的開銷，並且在減少延遲的同時增加了總體吞吐量。鏈式操作是可以配置的(後面出文章具體介紹)。

下圖dataflow的例子，使用五個subtasks執行，於是需要五個並發線程。

十，flink的角色

Flink運行時由兩種類型的進程組成：

1),JobManager也叫master協調分散式執行。他們調度任務，協調checkpoints，協調故障恢復等。至少有一個JobManager。高可用情況下可以啟動多個JobManager，其中一個選舉為leader，其餘為standby。

2),TaskManager也叫worker，負責執行具體的tasks。緩存，交換數據流。至少有一個TaskManager。

JobManager和TaskManager可以用很多種方式啟動：可以直接作為Standalone集群啟動，也可以被yarn或者Mesos管理。TaskManager連接到JobManager，宣布自己可用，並接受分配的工作。

客戶端不是runtime和程序執行的一部分，而是用於準備並發送數據流到JobManager。

之後，客戶端可以斷開連接或保持連接以接收進度報告。客戶端作為觸發執行的Java / Scala程序的一部分運行，或在命令行進程中運行./bin/flink運行。。

十一，Task Slots and Resources

每個worker(或者叫TaskManager)是一個jvm進程，可以在獨立的線程中執行一個或者多個subtasks。為了控制worker接受tasks數，worker稱之為任務槽數（至少有一個）。

每個task slot代表TaskManager的一個固定部分資源。例如，一個TaskManager有三個slot，會均分其管理的內存給每個slot。按槽分配資源意味著一個subtask不會與其它job的subtask競爭管理的內存，而是使用保留的內存資源。這裡並沒有實現cpu資源的隔離。僅僅實現了基於內存的資源隔離。

通過調整task slots的數量，用戶可以定義子任務彼此隔離的方式。每個TaskManager擁有一個slot意味著每個任務組在單獨的JVM中運行（例如，可以在單獨的容器中啟動）。擁有多個slots意味著更多的子任務共享相同的JVM。相同JVM中的任務共享TCP連接（通過復用）和心跳消息。他們還可以共享數據集和數據結構，從而減少每個任務的開銷。

默認情況下，Flink允許子任務共享slot，即使它們是不同tasks的subtasks，只要它們來自相同的job。結果是一個slot可以處理整個job pipeline。允許這個slot共享有兩個主要好處：

1),Flink集群需要與job中使用的最高並行度完全相同的task slot數。不需要計算一個程序總共包含多少任務（具有不同的並行性）。

2),更容易獲得更好的資源利用率。沒有slot共享，非密集的source / map()子任務將擁有與資源密集型窗口子任務一樣多的資源。使用slot共享，通過將例子中國的基礎並行度從2增加到6，可以充分利用slot資源，同時保證heavy subtasks均勻的分配到不同taskmanager中。

也有resource組機制，可以組織不希望的slot共享。

作為經驗法則，task slot最佳默認數量將是CPU內核的數量。

十二，State Backends

key/values索引存儲的確切數據結構取決於所選的State Backends(目前有三種：MemoryStateBackend，FsStateBackend，RocksDBStateBackend)。一種state backend將數據存儲於內存的hash map中，另一種state backend使用的是RocksDB，還有就是使用文件。除了定義保存狀態的數據結構之外，state backends還實現邏輯以獲取鍵/值狀態的point-in-time快照，並將該快照存儲為checkpoint的一部分。

十三，Savepoints

使用Data Stream API編寫的程序可以從Savepoints恢復執行。Savepoints允許更新程序和Flink集群，而不會丟失任何狀態。

Savepoints是手動觸發的checkpoints，它們記錄程序的快照並將其寫入狀態後端。他們依靠這個常規的檢查點機制。執行過程中，定期在工作節點上快照並生成檢查點。為了恢復，只需要最後完成的檢查點，一旦新的檢查點完成，可以安全地丟棄較舊的檢查點。Savepoints與這些定期checkpoints類似，除了它們由用戶觸發，並且在較新的檢查點完成時不會自動過期。可以從命令行創建保存點，也可以通過REST API取消作業。

十四，總結

Flink作為一個流式處理的框架，在實時計算中也是很常見的。Flink應用程序總共有以下三個部分：

1),Data source:Flink處理的輸入數據

2),Transformations：Flink修改傳入數據時的處理步驟

3),Data sink: flink處理結束後輸出位置

kafka，hbase，spark等源碼入門到深入，spark機器學習，大數據安全，大數據運維，請關注浪尖公眾號，看高質量文章。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！