Flink大數據分析實戰

1. Flink運行架構及原理 1.1 YARN集羣架構 YARN集羣總體上是經典的主／從(Master/Slave)架構，主要由ResourceManager、NodeManager、ApplicationMaster和Container等幾個組件構成。 一個應用程序所需的Container分爲兩類：運行ApplicationMaster的Container和運行各類Task的Container。前者是由ResourceManager向內部的資源調度器申請和啓動的，後者是由ApplicationMaster向ResourceManager申請的，並由ApplicationMaster請求NodeManager進行啓動。 總的來說，我們可以這樣認爲：ResourceManager管理整個集羣，NodeManager管理集羣中的單個節點，ApplicationMaster管理單個應用程序（集羣中可能同時有多個應用程序在運行，每個應用程序都有各自的ApplicationMaster）。 1.2 Flink Standalone架構 Flink中的每一個操作算子稱爲一個Task（任務），例如1.7.3節的單詞計數中使用的flatMap()算子、map()算子等。每個Task在一個JVM線程中執行。多個Task可以在同一個JVM進程中共享TCP連接（通過多路複用技術）和心跳信息。它們還可能共享數據集和數據結構，從而降低每個Task的開銷。Task是基本的工作單元，由Flink的Runtime來執行。Task正好封裝了一個算子或者算子鏈（見2.2.1節的任務鏈）的並行實例。 TaskManager爲了控制執行的Task數量，將計算資源（內存）劃分爲多個Task Slot（任務槽），每個Task Slot代表TaskManager的一份固定內存資源，Task則在Task Slot中執行。例如，具有3個Task Slot的TaskManager會將其管理的內存資源分成3等份給每個Task Slot。這種劃分方式可以更好地對多個Task進行內存隔離（沒有進行CPU隔離），使其不會競爭資源。 1.3 Flink On YARN的架構 2. Flink任務調度原理 由於每個SubTask只能在一個線程中執行，爲了能夠減少線程間切換和緩衝的開銷，在降低延遲的同時提高整體吞吐量，Flink可以將多個連續的SubTask鏈接成一個Task在一個線程中執行。這種將多個SubTask連在一起的方式稱爲任務鏈。 爲了充分利用計算資源，提高計算效率，可以增加算子的實例數（SubTask數量）。一個特定算子的SubTask數量稱爲該算子的並行度，且任意兩個算子的並行度之間是獨立的，不同算子可能擁有不同的並行度。 4種並行度級別的作用順序爲：算子級別>執行環境級別>客戶端級別>系統級別。 3. 重要概念 在Flink中，數據流或數據集被劃分成多個獨立的子集，這些子集分佈到了不同的節點上，而每一個子集稱爲分區(Partition)。 DataSet和DataStream數據集都是分佈式數據集，分佈式數據集是指：一個數據集存儲在不同的服務器節點上，每個節點存儲數據集的一部分。 4. 作業流程 用戶編寫應用程序代碼，並通過Flink客戶端提交作業。程序一般爲Java或Scala語言，調用Flink API構建邏輯數據流圖，然後轉爲作業圖JobGraph，並附加到StreamExecutionEnvironment中。代碼和相關配置文件被編譯打包，被提交到JobManager的Dispatcher，形成一個應用作業。 Dispatcher（JobManager的一個組件）接收到這個作業，啓動JobManager，JobManager負責本次作業的各項協調工作。 接下來JobManager向ResourceManager申請本次作業所需的資源。 JobManager將用戶作業中的作業圖JobGraph轉化爲並行化的物理執行圖，對作業並行處理並將其子任務分發部署到多個TaskManager上執行。每個作業的並行子任務將在Task Slot中執行。至此，一個Flink作業就開始執行了。 TaskManager在執行計算任務的過程中可能會與其他TaskManager交換數據，會使用相應的數據交換策略。同時，TaskManager也會將一些任務狀態信息反饋給JobManager，這些信息包括任務啓動、運行或終止的狀態、快照的元數據等。 5. Transformation數據轉換 map()算子接收一個函數作爲參數，並把這個函數應用於DataStream的每個元素，最後將函數的返回結果作爲結果DataStream中對應元素的值，即將DataStream的每個元素轉換成新的元素。與map()算子類似，但是每個傳入該函數func的DataStream元素會返回0到多個元素，最終會將返回的所有元素合併到一個DataStream。 通過函數filter對源DataStream的每個元素進行過濾，並返回一個新的DataStream。 keyBy()算子主要作用於元素類型是元組或數組的DataStream上。使用該算子可以將DataStream中的元素按照指定的key（指定的字段）進行分組，具有相同key的元素將進入同一個分區中（不進行聚合），並且不改變原來元素的數據結構。 reduce()算子主要作用於KeyedStream上，對KeyedStream數據流進行滾動聚合，即將當前元素與上一個聚合值進行合併，並且發射出新值。 connect()算子可以連接兩個數據流，並保持各自元素的數據類型不變，允許在兩個流之間共享狀態數據。與union()算子相比，二者的區別在於：1)union()要求多個數據流的數據類型必須相同，connect()允許多個數據流中的元素類型可以不同。2)union()可以合併多個數據流，但connect()只能連接兩個數據流。3)union()的執行結果是DataStream，而connect()的執行結果是ConnectedStreams；ConnectedStreams表示兩個（可能）不同數據類型的連接流，可以對兩個流的數據應用不同的處理方法，當一個流上的操作直接影響另一個流上的操作時，連接流非常有用。可以通過流之間的共享狀態對兩個流進行操作。 6. 分區策略 對於ForwardPartitioner，必須保證上下游算子並行度一致，否則會拋出異常。 全局分區策略將上游所有分區中的所有元素髮送到了下游編號爲0的分區中。 Key分區策略根據元素Key的Hash值輸出到下游算子指定的實例。keyBy()算子底層正是使用的該分區策略，底層最終會調用KeyGroupStreamPartitioner的selectChannel()方法，計算每個Key對應的通道索引（通道編號，可理解爲分區編號），根據通道索引將Key發送到下游相應的分區中。 如果想將元素均勻地輸出到下游算子的每個實例，以實現負載均衡，同時又不希望使用平衡分區策略的全局負載均衡，則可以使用重新調節分區策略。該策略會盡可能避免數據在網絡間傳輸，而能否避免還取決於TaskManager的Task Slot數量、上下游算子的並行度等。 7. 窗口計算 事件時間(Event Time)：每個事件或元素在其生產設備上產生的時間。該時間通常在它們進入Flink之前嵌入事件中，並且可以從每個事件中提取事件時間戳。流程序的時間默認使用的是事件時間。 Flink的窗口可以分爲滾動窗口、滑動窗口、會話窗口、全局窗口，且每種窗口又可分別根據事件時間和處理時間進行創建。 使用滑動窗口時，需要設置窗口大小和滑動步長兩個參數。滑動步長決定了Flink以多大的頻率來創建新的窗口，步長較小，窗口的個數會很多。步長小於窗口的大小時，相鄰窗口會重疊，一個事件會被分配到多個窗口；步長大於窗口大小時，有些事件可能會丟失。 ReduceFunction指定如何聚合輸入中的兩個元素以產生相同類型的輸出元素。 AggregateFunction是聚合函數的基本接口，也是ReduceFunction的通用版本。與ReduceFunction相同，Flink將在窗口輸入元素到達時對其進行增量聚合。 使用ProcessWindowFunction可以獲得一個包含窗口所有元素的可迭代對象(Iterable)，以及一個可以訪問時間和狀態信息的上下文對象(Context)，這使得它比其他窗口函數提供了更多的靈活性。 由於ProcessWindowFunction是全量計算函數，如果既要獲得窗口信息又要進行增量聚合，則可以將ProcessWindowFunction與ReduceFunction或AggregateFunction結合使用。 Flink不保證窗口內元素的順序。這意味着，雖然清除器可以從窗口的開頭移除元素，但這些元素不一定是最先到達的。 8. 水印 爲了保證計算結果的正確性，需要讓窗口等待延遲數據到達後再進行計算，但是不能無限期地等待下去，必須有一種機制來確定何時觸發窗口計算，這種機制就是水印(Watermark)。 水印=進入Flink的當前最大事件時間‒允許最大延遲時間，當水印>=窗口結束時間時，立即觸發窗口計算，計算完畢後發射出計算結果並銷燬窗口；否則窗口將一直等待。 對於一些使用水印解決不了的、更爲嚴重的延遲數據，Flink默認是丟棄的，爲了保證數據不丟失，Flink提供了允許延遲(Allowed Lateness)和側道輸出機制(Side Output)。 由於下游算子對水印的計算方式是取所有不同的上游並行算子中水印的最小值，如果上游某一個算子因爲缺少數據遲遲沒有生成水印，就會出現事件時間傾斜問題，水印將不會發生變化，導致下游不能觸發計算。 爲了解決這個問題，可以使用WatermarkStrategy接口中已經實現的withIdleness(Duration idleTimeout)方法來檢測空閒輸入並將其標記爲空閒狀態。 9. 狀態管理 9.1 總結來說，Flink狀態管理的主要特性如下： 本地性：Flink狀態是存儲在使用它的機器本地的，並且可以內存訪問速度來獲取。 持久性：Flink狀態是容錯的，例如它可以自動按一定的時間間隔產生快照，並且在任務失敗後進行恢復。· 縱向可擴展性：Flink狀態可以存儲在集成的RocksDB（一種KV型數據庫）實例中，這種方式下可以通過增加本地磁盤來擴展空間。· 橫向可擴展性：Flink狀態可以隨着集羣的擴／縮容重新分佈。· 可查詢性：Flink狀態可以通過使用狀態查詢API從外部進行查詢。 9.2 AggregatingState<IN, OUT>：保留一個單值。和ReducingState不同的是，聚合類型可能與添加到狀態的元素的類型不同。使用add(IN)添加元素，並與已有的元素使用AggregateFunction進行聚合。 10. 容錯機制◆ Flink將Checkpoint快照的存儲位置稱爲狀態後端(State Backend)。狀態後端有兩種實現：一種基於RocksDB將工作狀態保存在磁盤上，使用RocksDBStateBackend類型；另一種基於堆將工作狀態保存在Java的堆內存中。基於堆的狀態後端有兩種類型：FsStateBackend和MemoryStateBackend。FsStateBackend會定時將其狀態快照持久化到分佈式文件系統中，MemoryStateBackend使用JobManager的堆內存保存狀態快照。 ◆ RocksDBStateBackend適用於狀態非常大、窗口非常長的作業以及所有高可用的場景。由於其允許存儲非常大的狀態，因此意味着使用RocksDBStateBackend將會使應用程序的最大吞吐量降低。所有的讀寫都必須序列化、反序列化操作，比基於堆內存的狀態後端的效率要低很多。 ◆ FsStateBackend狀態後端適用於狀態比較大、窗口比較長的作業以及所有高可用的場景。對於一些以分鐘爲單位的窗口聚合，建議使用該狀態後端。 ◆ 通常，建議在生產中避免使用MemoryStateBackend，因爲它將快照存儲在JobManager的內存中，而不是持久化到磁盤。當需要在FsStateBackend和RocksDBStateBackend之間進行選擇時，需要從性能和可伸縮性方面進行考慮。FsStateBackend非常快，因爲每個狀態訪問和更新操作在Java堆內存上，但是狀態大小受集羣內可用內存的限制。另一方面，RocksDBStateBackend可以根據可用磁盤空間進行擴展，並且是唯一支持增量快照的狀態後端。但是每個狀態訪問和更新都需要序列化／反序列化，這樣會導致平均性能比內存狀態後端慢一個數量級。 ◆ Savepoint（保存點）是用戶手動觸發的Checkpoint，由用戶手動創建、擁有和刪除，它獲取狀態的快照並將其寫入狀態後端。Savepoint主要用於手動的狀態數據備份和恢復，常用於在升級和維護集羣的過程中保存狀態數據，避免系統無法恢復到原有的計算狀態。例如，需要升級程序之前先執行一次Savepoint，升級後可以繼續從升級前的那個點進行計算，保證數據不中斷。 11. SQL API TUMBLE函數將每個元素分配給一個具有指定窗口大小的滾動窗口。滾動窗口有一個固定的大小，不重疊。 ◆ HOP窗口函數將元素分配給固定長度的窗口。與TUMBLE窗口函數一樣，窗口的大小是由窗口大小參數配置的。另一個滑動步伐參數控制HOP窗口的啓動頻率（即間隔多長時間開啓一個新窗口）。因此，如果滑動步伐小於窗口大小，HOP窗口可以重疊。HOP窗口也被稱爲“滑動窗口”。 ◆ CUMULATE窗口在某些情況下是非常有用的，例如在固定的窗口間隔內觸發滾動窗口。 ◆ 窗口聚合定義在GROUP BY子句中，該子句包含應用於窗口函數的window_start和window_end列。與使用常規GROUP BY子句的查詢一樣，使用按窗口分組聚合的查詢將爲每個組計算單個結果行。 ◆ 窗口聚合也支持GROUPING SETS語法。該語法允許比標準GROUP BY所描述的更復雜的分組操作。其可以對GROUP BY分組後的每一組繼續按照指定的列進行分組，併爲每個組計算聚合，就像簡單的GROUP BY子句一樣。 ◆ ROLLUP是一種用於指定通用類型分組集的簡寫符號。它表示給定的表達式列表和列表的所有前綴，包括空列表。使用ROLLUP的窗口聚合要求window_start和window_end列都必須在GROUP BY子句中，而不是在ROLLUP子句中。 ◆ CUBE是一種用於指定通用類型分組集的簡寫符號，它表示給定的列表及其所有可能的子集。 ◆ OVER聚合計算一段有序行範圍內的每個輸入行的聚合值。 12. 流圖&執行圖 1)Flink Client將作業的應用程序代碼生成StreamGraph（在批模式下生成的是OptimizedPlan，這裏暫不講解）。StreamGraph是表示流處理程序拓撲的數據結構，描述了算子和算子之間邏輯上的拓撲關係，封裝了生成作業圖(JobGraph)的必要信息。 2)將StreamGraph轉換爲JobGraph。JobGraph表示JobManager可接受的低級別Flink數據流程序。所有來自更高級別API的程序都被轉換成JobGraph。JobGraph是一個頂點和中間結果的圖，這些頂點和中間結果連接在一起形成一個DAG（有向無環圖）。JobGraph定義了作業範圍的配置信息，而每個頂點和中間結果定義了具體操作和中間數據的特徵。3)將JobGraph提交給Dispatcher。Dispatcher組件負責接收作業提交、在故障時恢復作業、監控Flink會話集羣的狀態等。4)Dispatcher根據JobGraph創建相應的JobManager並運行。5)JobManager將JobGraph轉換爲ExecutionGraph。ExecutionGraph是協調數據流的分佈式執行的中心數據結構，它保留了每個並行任務、每個中間流以及它們之間的通信信息。6)JobManager將ExecutionGraph轉換爲物理執行圖。 ◆ StreamNode和StreamEdge之間有着雙向的依賴關係。StreamEdge包含其連接的源節點（源StreamNode，使用sourceVertex屬性表示）和目的節點（目的StreamNode，使用sourceVertex屬性表示） ◆ StreamGraph和JobGraph的轉化生成都是在Flink客戶端，而最終Flink作業運行時調度層的核心執行圖ExecutionGraph是在服務端的JobManager中生成的。 ◆ 總的來說，ExecutionGraph的轉換過程爲：將JobGraph按照拓撲排序後得到一個JobVertex集合，遍歷該JobVertex集合，即從Source開始，將JobVertex封裝成ExecutionJobVertex，並依次創建ExecutionVertex、Execution、IntermediateResult和IntermediateResultPartition。然後通過ejv.connectToPredecessor()方法創建ExecutionEdge，建立當前節點與其上游節點之間的聯繫，即連接ExecutionVertex和IntermediateResultPartition。構建好ExecutionGraph後，接下來會基於ExecutionGraph觸發作業的調度，申請Task Slot，部署任務到TaskManager執行。

我是本書作者，作者微信jock90，可與我交流疑難問題，也可加入讀者羣相互學習。

推薦給初學者，常用使用場景都覆蓋了。