一、flink--架構、運行、調度原理

一、flink概述

1.1 流處理技術語義

At most once（最多一次）：每條數據記錄最多被處理一次，潛臺詞也表明數據會有丟失（沒被處理掉）的可能。

At least once（最少一次）：每條數據記錄至少被處理一次。這個比上一點強的地方在于這里至少保證數據不會丟，至少被處理過，唯一不足之處在于數據可能會被重復處理。

Exactly once（恰好一次）：每條數據記錄正好被處理一次。沒有數據丟失，也沒有重復的數據處理。這一點是3個語義里要求最高的。

1.2 flink是什么

? Flink主頁在其頂部展示了該項目的理念：“Apache Flink是為分布式、高性能、隨時可用以及準確的流處理應用程序打造的開源流處理框架”。Apache Flink是一個框架和分布式處理引擎，用于對***和有界數據流進行有狀態計算。Flink被設計在所有常見的集群環境中運行，以內存執行速度和任意規模來執行計算。

1.3 flink基本框架

? 批處理的特點是有界、持久、大量，批處理非常適合需要訪問全套記錄才能完成的計算工作，一般用于離線統計。流處理的特點是***、實時，流處理方式無需針對整個數據集執行操作，而是對通過系統傳輸的每個數據項執行操作，一般用于實時統計。
在Spark生態體系中，對于批處理和流處理采用了不同的技術框架，批處理由SparkSQL實現，流處理由Spark Streaming實現，這也是大部分框架采用的策略，使用獨立的處理器實現批處理和流處理，而Flink可以同時實現批處理和流處理。
? Flink是如何同時實現批處理與流處理的呢？答案是，Flink將批處理（即處理有限的靜態數據）視作一種特殊的流處理。
? Flink的核心計算架構是下圖中的Flink Runtime執行引擎，它是一個分布式系統，能夠接受數據流程序并在一臺或多臺機器上以容錯方式執行。
? Flink Runtime執行引擎可以作為YARN（Yet Another Resource Negotiator）的應用程序在集群上運行，也可以在Mesos集群上運行，還可以在單機上運行（這對于調試Flink應用程序來說非常有用）。

? 圖1.1 flink--基本架構

? 上圖為Flink技術棧的核心組成部分，值得一提的是，Flink分別提供了面向流式處理的接口（DataStream API）和面向批處理的接口（DataSet API）。因此，Flink既可以完成流處理，也可以完成批處理。Flink支持的拓展庫涉及機器學習（FlinkML）、復雜事件處理（CEP）、以及圖計算（Gelly），還有分別針對流處理和批處理的Table API。
? 能被Flink Runtime執行引擎接受的程序很強大，但是這樣的程序有著冗長的代碼，編寫起來也很費力，基于這個原因，Flink提供了封裝在Runtime執行引擎之上的API，以幫助用戶方便地生成流式計算程序。Flink 提供了用于流處理的DataStream API和用于批處理的DataSet API。值得注意的是，盡管Flink Runtime執行引擎是基于流處理的，但是DataSet API先于DataStream API被開發出來，這是因為工業界對無限流處理的需求在Flink誕生之初并不大。
? DataStream API可以流暢地分析無限數據流，并且可以用Java或者Scala來實現。開發人員需要基于一個叫DataStream的數據結構來開發，這個數據結構用于表示永不停止的分布式數據流。
? Flink的分布式特點體現在它能夠在成百上千臺機器上運行，它將大型的計算任務分成許多小的部分，每個機器執行一部分。Flink能夠自動地確保發生機器故障或者其他錯誤時計算能夠持續進行，或者在修復bug或進行版本升級后有計劃地再執行一次。這種能力使得開發人員不需要擔心運行失敗。Flink本質上使用容錯性數據流，這使得開發人員可以分析持續生成且永遠不結束的數據（即流處理）。

1.4 無窮數據流和有限數據流

無窮數據集：無窮的持續集合的數據集合
有限數據集：有限不會改變的數據集合

常見的無窮數據集合有：
用戶與客戶端的實時交互數據
應用實時產生的日志
金融市場的實時交易記錄

1.5 Flink和storm對比

1.6 flink特性

1、高吞吐和低延遲性

2、支持 Event Time 和亂序事件
Flink 支持了流處理和 Event Time 語義的窗口機制。
Event time 使得計算亂序到達的事件或可能延遲到達的事件更加簡單。

3、狀態計算的 exactly-once 語義
故障狀態下，需要重啟計算任務，這時候需要避免已經處理過的數據的重復處理。
流程序可以在計算過程中維護自定義狀態。
Flink 的 checkpointing 機制保證了即時在故障發生下也能保障狀態的 exactly once 語義。

4、高度靈活的流式窗口
Flink 支持在時間窗口，統計窗口，session 窗口，以及數據驅動的窗口
窗口可以通過靈活的觸發條件來定制，以支持復雜的流計算模式。

5、帶反壓的連續流模型
數據流應用執行的是不間斷的（常駐）operators。
Flink streaming 在運行時有著天然的流控：慢的數據 sink 節點會反壓（backpressure）快的數據源（sources）。

6、容錯性
Flink 的容錯機制是基于 Chandy-Lamport distributed snapshots 來實現的。
這種機制是非常輕量級的，允許系統擁有高吞吐率的同時還能提供強一致性的保障。

7、Batch 和 Streaming 一個系統流處理和批處理共用一個引擎
Flink 為流處理和批處理應用公用一個通用的引擎。批處理應用可以以一種特殊的流處理應用高效地運行。

8、內存管理
Flink 在 JVM 中實現了自己的內存管理。
應用可以超出主內存的大小限制，并且承受更少的垃圾收集的開銷。

9、迭代和增量迭代
Flink 具有迭代計算的專門支持（比如在機器學習和圖計算中）。
增量迭代可以利用依賴計算來更快地收斂。

10、程序調優
批處理程序會自動地優化一些場景，比如避免一些昂貴的操作（如 shuffles 和 sorts），還有緩存一些中間數據。

1.7 flink應用場景

? Apache Flink 功能強大，支持開發和運行多種不同種類的應用程序。它的主要特性包括：批流一體化、精密的狀態管理、事件時間支持以及精確一次的狀態一致性保障等。Flink 不僅可以運行在包括 YARN、 Mesos、Kubernetes 在內的多種資源管理框架上，還支持在裸機集群上獨立部署。在啟用高可用選項的情況下，它不存在單點失效問題。事實證明，Flink 已經可以擴展到數千核心，其狀態可以達到 TB 級別，且仍能保持高吞吐、低延遲的特性。世界各地有很多要求嚴苛的流處理應用都運行在 Flink 之上。

1.7.1 事件驅動型應用

反欺詐
異常檢測
基于規則的報警
業務流程監控
Web應用

1.7.2 數據分析應用

電信網路質量監控
移動應用中的產品更新及實驗評估分析
大規模圖分析

1.7.3 數據管道應用

電子商務中的實時查詢索引構建
電子商務中的持續ETL

二、Flink基本架構

2.1 flink中的角色

Flink運行時包含了兩種類型的處理器：
JobManager處理器：也稱之為Master，用于協調分布式執行，它們用來調度task，協調檢查點，協調失敗時恢復等。Flink運行時至少存在一個master處理器，如果配置高可用模式則會存在多個master處理器，它們其中有一個是leader，而其他的都是standby。

TaskManager處理器：也稱之為Worker，用于執行一個dataflow的task(或者特殊的subtask)、數據緩沖和data stream的交換，Flink運行時至少會存在一個worker處理器。

? 圖2.1 flink--JobManager與TaskManager

Master和Worker處理器可以直接在物理機上啟動，或者通過像YARN這樣的資源調度框架。Worker連接到Master，告知自身的可用性進而獲得任務分配。

2.2 ***數據流與有界數據流

***數據流：
***數據流有一個開始但是沒有結束，它們不會在生成時終止并提供數據，必須連續處理***流，也就是說必須在獲取后立即處理event。對于***數據流我們無法等待所有數據都到達，因為輸入是***的，并且在任何時間點都不會完成。處理***數據通常要求以特定順序（例如事件發生的順序）獲取event，以便能夠推斷結果完整性。

有界數據流：
有界數據流有明確定義的開始和結束，可以在執行任何計算之前通過獲取所有數據來處理有界流，處理有界流不需要有序獲取，因為可以始終對有界數據集進行排序，有界流的處理也稱為批處理。

? Apache Flink是一個面向分布式數據流處理和批量數據處理的開源計算平臺，它能夠基于同一個Flink運行時(Flink Runtime)，提供支持流處理和批處理兩種類型應用的功能。現有的開源計算方案，會把流處理和批處理作為兩種不同的應用類型，因為它們要實現的目標是完全不相同的：流處理一般需要支持低延遲、Exactly-once保證，而批處理需要支持高吞吐、高效處理，所以在實現的時候通常是分別給出兩套實現方法，或者通過一個獨立的開源框架來實現其中每一種處理方案。例如，實現批處理的開源方案有MapReduce、Tez、Crunch、Spark，實現流處理的開源方案有Samza、Storm。
Flink在實現流處理和批處理時，與傳統的一些方案完全不同，它從另一個視角看待流處理和批處理，將二者統一起來：Flink是完全支持流處理，也就是說作為流處理看待時輸入數據流是***的；批處理被作為一種特殊的流處理，只是它的輸入數據流被定義為有界的。基于同一個Flink運行時(Flink Runtime)，分別提供了流處理和批處理API，而這兩種API也是實現上層面向流處理、批處理類型應用框架的基礎。

2.3 flink數據流編程接口抽象

Flink提供了不同級別的抽象，以開發流或批處理作業，如下圖所示：

? 圖2.3 flink編程接口抽象
? 最底層級的抽象僅僅提供了有狀態流，它將通過過程函數（Process Function）被嵌入到DataStream API中。底層過程函數（Process Function） 與 DataStream API 相集成，使其可以對某些特定的操作進行底層的抽象，它允許用戶可以自由地處理來自一個或多個數據流的事件，并使用一致的容錯的狀態。除此之外，用戶可以注冊事件時間并處理時間回調，從而使程序可以處理復雜的計算。
? 實際上，大多數應用并不需要上述的底層抽象，而是針對核心API（Core APIs） 進行編程，比如DataStream API（有界或***流數據）以及DataSet API（有界數據集）。這些API為數據處理提供了通用的構建模塊，比如由用戶定義的多種形式的轉換（transformations），連接（joins），聚合（aggregations），窗口操作（windows）等等。DataSet API 為有界數據集提供了額外的支持，例如循環與迭代。這些API處理的數據類型以類（classes）的形式由各自的編程語言所表示。
? Table API 是以表為中心的聲明式編程，其中表可能會動態變化（在表達流數據時）。Table API遵循（擴展的）關系模型：表有二維數據結構（schema）（類似于關系數據庫中的表），同時API提供可比較的操作，例如select、project、join、group-by、aggregate等。Table API程序聲明式地定義了什么邏輯操作應該執行，而不是準確地確定這些操作代碼的看上去如何 。 盡管Table API可以通過多種類型的用戶自定義函數（UDF）進行擴展，其仍不如核心API更具表達能力，但是使用起來卻更加簡潔（代碼量更少）。除此之外，Table API程序在執行之前會經過內置優化器進行優化。
你可以在表與 DataStream/DataSet 之間無縫切換，以允許程序將 Table API 與 DataStream 以及 DataSet 混合使用。
? Flink提供的最高層級的抽象是 SQL 。這一層抽象在語法與表達能力上與 Table API 類似，但是是以SQL查詢表達式的形式表現程序。SQL抽象與Table API交互密切，同時SQL查詢可以直接在Table API定義的表上執行。

三、flink運行架構

3.1 提交任務到yarn的流程

flink在生產中，一般是使用yarn作為資源調度平臺，比較少使用standalone的方式進行資源調度。所以這里以yarn為例，說明flink提交任務到yarn的流程。

? 圖3.1 flink--提交任務到yarn流程
? Flink任務提交后，Client向HDFS上傳Flink的Jar包和配置，之后向Yarn ResourceManager提交任務，ResourceManager分配Container資源并通知對應的NodeManager啟動ApplicationMaster，ApplicationMaster啟動后加載Flink的Jar包和配置構建環境，然后啟動JobManager，之后ApplicationMaster向ResourceManager申請資源啟動TaskManager，ResourceManager分配Container資源后，由ApplicationMaster通知資源所在節點的NodeManager啟動TaskManager，NodeManager加載Flink的Jar包和配置構建環境并啟動TaskManager，TaskManager啟動后向JobManager發送心跳包，并等待JobManager向其分配任務。

3.2 任務調度組件

? 圖3.2 flink--任務調度
1、 Program Code：我們編寫的 Flink 應用程序代碼

2、 Job Client：Job Client 不是 Flink 程序執行的內部部分，但它是任務執行的起點。 Job Client 負責接受用戶的程序代碼，然后創建數據流，將數據流提交給 Job Manager 以便進一步執行。 執行完成后，Job Client 將結果返回給用戶

3、 JobManager：主進程（也稱為作業管理器）協調和管理程序的執行。 它的主要職責包括安排任務，管理checkpoint ，故障恢復等。機器集群中至少要有一個 master，master 負責調度 task，協調 checkpoints 和容災，高可用設置的話可以有多個 master，但要保證一個是active, 其他是 standby; Job Manager 包含 Actor system(通信系統)、Scheduler（調度）、Check pointing 三個重要的組件

4、 Task Manager：從 Job Manager 處接收需要部署的 Task。Task Manager 是在 JVM 中的一個或多個線程中執行任務的工作節點。 任務執行的并行性由每個 Task Manager 上可用的任務槽（task slot）決定。 每個任務代表分配給任務槽的一組資源。 例如，如果 Task Manager 有四個插槽，那么它將為每個插槽分配 25％ 的內存。 可以在任務槽中運行一個或多個線程。 同一插槽中的線程共享相同的 JVM。 同一 JVM 中的任務共享 TCP 連接和心跳消息。Task Manager 的一個 Slot 代表一個可用線程，該線程具有固定的內存，注意 Slot 只對內存隔離，沒有對 CPU 隔離。默認情況下，Flink 允許子任務共享 Slot，即使它們是不同 task 的 subtask，只要它們來自相同的 job。這種共享可以有更好的資源利用率。

3.3 TaskManager和slots原理

? 每一個worker(TaskManager)是一個JVM進程，它可能會在獨立的線程上執行一個或多個subtask。為了控制一個worker能接收多少個task，worker通過task slot來進行控制（一個worker至少有一個task slot）。
? 每個task slot表示TaskManager擁有資源的一個固定大小的子集。假如一個TaskManager有三個slot，那么它會將其管理的內存平均分成三份給各個slot。資源slot化意味著一個subtask將不需要跟來自其他job的subtask競爭被管理的內存，取而代之的是它將擁有一定數量的內存儲備。需要注意的是，這里不會涉及到CPU的隔離，slot目前僅僅用來隔離task的受管理的內存。
? 通過調整task slot的數量，允許用戶定義subtask之間如何互相隔離。如果一個TaskManager一個slot，那將意味著每個task group運行在獨立的JVM中（該JVM可能是通過一個特定的容器啟動的），而一個TaskManager多個slot意味著更多的subtask可以共享同一個JVM。而在同一個JVM進程中的task將共享TCP連接（基于多路復用）和心跳消息。它們也可能共享數據集和數據結構，因此這減少了每個task的負載。

? 圖3.3 taskManager和slots
? Task Slot是靜態的概念，是指TaskManager具有的并發執行能力，可以通過參數taskmanager.numberOfTaskSlots進行配置，而并行度parallelism是動態概念，即TaskManager運行程序時實際使用的并發能力，可以通過參數parallelism.default進行配置。
? 也就是說，假設一共有3個TaskManager，每一個TaskManager中的分配3個TaskSlot，也就是每個TaskManager可以接收3個task，一共9個TaskSlot，如果我們設置parallelism.default=1，即運行程序默認的并行度為1，9個TaskSlot只用了1個，有8個空閑，因此，設置合適的并行度才能提高效率。實際上slots限制的限制了該taskmanager在整個集群中能夠并行運行task的數目，而parallelism.default則是限制單個job能夠使用slot的數量，但是允許多個job同時運行，所以實際上是對單個job的并發限制。

3.4 程序與數據流

? Flink程序的基礎構建模塊是 流（streams） 與 轉換（transformations）（需要注意的是，Flink的DataSet API所使用的DataSets其內部也是stream）。一個stream可以看成一個中間結果，而一個transformations是以一個或多個stream作為輸入的某種operation，該operation利用這些stream進行計算從而產生一個或多個result stream。
? 在運行時，Flink上運行的程序會被映射成streaming dataflows，它包含了streams和transformations operators。每一個dataflow以一個或多個sources開始以一個或多個sinks結束。dataflow類似于任意的有向無環圖（DAG），當然特定形式的環可以通過iteration構建。在大部分情況下，程序中的transformations跟dataflow中的operator是一一對應的關系，但有時候，一個transformation可能對應多個operator。

? 圖3.4 程序與數據流

3.5 并行數據流（operator并行）

? Flink程序的執行具有并行、分布式的特性。在執行過程中，一個 stream 包含一個或多個 stream partition ，而每一個 operator 包含一個或多個 operator subtask，這些operator subtasks在不同的線程、不同的物理機或不同的容器中彼此互不依賴得執行。
? 一個特定operator的subtask的個數被稱之為其parallelism(并行度)。一個stream的并行度總是等同于其producing operator的并行度。一個程序中，不同的operator可能具有不同的并行度。

? 圖3.5 并行數據流
? Stream在operator之間傳輸數據的形式可以是one-to-one(forwarding)的模式也可以是redistributing的模式，具體是哪一種形式，取決于operator的種類。
? One-to-one：stream(比如在source和map operator之間)維護著分區以及元素的順序。那意味著map operator的subtask看到的元素的個數以及順序跟source operator的subtask生產的元素的個數、順序相同，map、fliter、flatMap等算子都是one-to-one的對應關系。不會改變分區的情況下，才能是該模式。
? Redistributing：stream(map()跟keyBy/window之間或者keyBy/window跟sink之間)的分區會發生改變。每一個operator subtask依據所選擇的transformation發送數據到不同的目標subtask。例如，keyBy() 基于hashCode重分區、broadcast和rebalance會隨機重新分區，這些算子都會引起redistribute過程，而redistribute過程就類似于Spark中的shuffle過程。

3.6 task和operator chains

? 出于分布式執行的目的，Flink將同一類operator的subtask鏈接在一起形成task，每個task在一個線程中執行。將operators鏈接成task是非常有效的優化：它能減少線程之間的切換和基于緩存區的數據交換，在減少時延的同時提升吞吐量。鏈接的行為可以在編程API中進行指定。這個task鏈的方式其實和spark中的劃分stage，然后構建task鏈的方式是一模一樣的，如果理解不了，可以看之前spark的文章。
下面這幅圖，展示了5個subtask以5個并行的線程來執行：

? 圖3.6 flink--operator chains
看到上面的圖，因為keyBy這個算子是會導致重分區的，那么以這里為界限，劃分stage，然后前面的 source 和map可以獨立構建task鏈，后面的keyBy、window另外構建task鏈。加上最后統一的sink操作，其實是5個task鏈，然后根據先后順序運行。這個機制和spark的一樣。