Flink中設計了用戶自定義函數體系(User Defined Function,UDF),開發(fā)人員實現業(yè)務邏輯就是開發(fā)UDF。

一、環(huán)境對象

StreamExecutionEnvironment是Flink應用開發(fā)時的概念，表示流計算作業(yè)的執(zhí)行環(huán)境，是作業(yè)開發(fā)的入口、數據源接口、生成和轉換DataStream的接口、數據Sink的接口、作業(yè)配置接口、作業(yè)啟動執(zhí)行的入口。

Environment是運行時作業(yè)級別的概念，從StreamExecutionEnvironment中的配置信息衍生而來。進入到Flink作業(yè)執(zhí)行的時刻，作業(yè)需要的是相關的配置信息，如作業(yè)的名稱、并行度、作業(yè)編號JobID、監(jiān)控的Metric、容錯的配置信息、IO等，用StreamExecutionRuntime對象就不適合了，很多API是不需要的，所以在Flink中抽象出了Environment作為運行時刻的上下文信息。

RuntimeContext是運行時Task實例級別的概念。Environment本身仍然是比較粗粒度作業(yè)級別的配置，對于每一個Task而言，其本身有更細節(jié)的配置信息,所以Flink又抽象了RuntimeContext,每一個Task實例有自己的RuntimeContext。

環(huán)境對象關系如下：

1.1 執(zhí)行環(huán)境

StreamExecutionEnvironment

Flink流計算應用的執(zhí)行環(huán)境，是Flink作業(yè)開發(fā)和啟動執(zhí)行的入口

開發(fā)者對StreamExecutionEnvironment的實現是無感知的。

LocalStreamEnvironment

本地執(zhí)行環(huán)境，在單個JVM中使用多線程模擬Flink集群。

其基本的工作流程如下：

1) 執(zhí)行Flink作業(yè)的Main函數生成Streamgraph,轉化為JobGraph。

2) 設置任務運行的配置信息。

3) 根據配置信息啟動對應的LocalFlinkMiniCluster。

4) 根據配置信息和miniCluster生成對應的MiniClusterClient。

5) 通過MiniClusterClient提交JobGraph 到MiniCluster。

RemoteStreamEnvironment

在大規(guī)模數據中心中部署的Flink生產集群的執(zhí)行環(huán)境。

當將作業(yè)發(fā)布到Flink集群的時候，使用RemoteStreamEnvironment。

其基本的工作流程如下：

1) 執(zhí)行Flink作業(yè)的Main函數生成Streamgraph，轉化為JobGraph。

2) 設置任務運行的配置信息。

3) 提交JobGraph到遠程的Flink集群。

StreamContextEnvironment

在Cli命令行或者單元測試時候會被使用，執(zhí)行步驟同上。

StreamPlanEnvironment

在Flink Web UI管理界面中可視化展現Job的時候，專門用來生成執(zhí)行計劃(實際上就是StreamGraph)

ScalaShellStreamEnvironment

這是Scala Shell執(zhí)行環(huán)境，可以在命令行中交互式開發(fā)Flink作業(yè)。

其基本工作流程如下：

• 校驗部署模式，目前Scala Shell僅支持attached模式。
• 上傳每個作業(yè)需要的Jar文件。

其余步驟與RemoteStreamEnvironment類似。

1.2 運行時環(huán)境

RuntimeEnvironment

在Task開始執(zhí)行時進行初始化，把Task運行相關的信息都封裝到該對象中，其中不光包含了配置信息，運行時的各種服務也會被包裝到其中。

SavepointEnvironment

SavepointEnvironment是Environment的最小化實現，在狀態(tài)處理器的API中使用。

1.3 運行時上下文

RuntimeContext是Function運行時的上下文，封裝了Function運行時可能需要的所有信息，讓Function在運行時能夠獲取到作業(yè)級別的信息，如并行度相關信息、Task名稱、執(zhí)行配置信息(ExecutionConfig)、State等。

Function的每個實例都有一個RuntimeContext對象，在RichFunction中通過getRunctionContext()可以訪問該對象。

RuntimeContext的類體系圖如下:

StreamingRuntimeContext：

在流計算UDF中使用的上下文，用來訪問作業(yè)信息、狀態(tài)等。

DistributedRuntimeUDFContext：

由運行時UDF所在的批處理算子創(chuàng)建，在DataSet批處理中使用。

RuntimeUDFContext：                                                          

在批處理應用的UDF中使用。

SavepointRuntimeContext：

支持對檢查點和保存點進行操作，包括讀取、變更、寫入等。

CepRuntimeContext：

CEP復雜事件處理中使用的上下文。

二、數據流元素

數據流元素在Flink中叫做StreamElement

• 有數據記錄StreamRecord,
• 延遲標記LatencyMarker、Watermark、
• 流狀態(tài)標記StreamStatus這四種。

在執(zhí)行層面，4種數據流元素都被序列化成二進制數據，形成混合的數據流，在算子中將混合數據流中的數據流元素反序列化出來。

StreamRecord

StreamRecord表示數據流中的一條記錄(或者叫做一個事件)，也叫數據記錄。

包含以下內容：

1）數據的值本身

2）時間戳(可選)

LatencyMarker

用來近似評估延遲，LatencyMarker在Source中創(chuàng)建，并向下游發(fā)送，繞過業(yè)務處理邏輯，在Sink節(jié)點中使用LatencyMarker估計數據在整個DAG圖中的流轉花費的時間。

LatencyMarker包含信息如下：

1）周期性的在數據源算子中創(chuàng)造出來的時間戳。

2）算子編號

3）數據源算子所在的Task編號

Watermark

是一個時間戳，用來告訴算子所有時間早于等于Watermark的事件或記錄都已經達到，不會再有比Watermark更早的記錄。

StreamStatus

用來通知Task是否會繼續(xù)接收到上游的記錄或者Watermark。在數據源算子中生成，向下游沿著DataFlow傳遞。

有兩種表示狀態(tài):

1）空閑狀態(tài)(IDLE)

2）活動狀態(tài)(ACTIVE)

三、數據轉換

數據轉換在Flink中叫做Transformation,是銜接DataStream Api和Flink內核的邏輯結構。

Transformation有兩大類：

1）物理Transformation: 會轉換成算子，繼承了PhysicalTransformation。

2）虛擬Transformation: 不會轉換成具體算子。

Tranformation包含了Flink的運行時關鍵參數:

1）name：轉換器名稱，主要用于可視化。

2）uid：用戶指定的uid,該uid的主要目的是在job重啟時再次分配跟之前相同的uid,可以持久保存狀態(tài)。

3）bufferTimeout：buffer超時時間。

4）parallelism：并行度。

5）id：跟屬性uid無關，生成方式是基于一個靜態(tài)累加器。

6）outputType：輸出類型，用來進行序列化數據。

7）slotSharingGroup：給當前的Transformation設置Slot共享組。

3.1 物理Transformation SourceTransformation

從數據源讀取數據的Transformation，是Flink作業(yè)的起點。

只有下游Transformation，沒有上游輸入。

SinkTransformation

將數據寫到外部存儲的Transformation,是Flink作業(yè)的終點。

OneInputTransformation

單流輸入的Transformation(只接收一個輸入流)，跟上面的SinkTransformation構造器類似，同樣需要input和operator參數。

TwoInputTransformation

雙輸入的Transformation(接收兩種流作為輸入)，分別叫做第一輸入和第二輸入。

3.2 虛擬Transformation SideOutputTransformation

在旁路輸出中轉換而來，表示上游Transformation的一個分流。

每個sideoutput通過OutputTag標識。

SplitTransformation

用來按條件切分數據流，該轉換用于將一個流拆分成多個流。

SelectTransformation

與SplitTransformation配合使用，用來在下游選擇SplitTransformation切分的數據流。

PartitionTransformation

該轉換器用于改變輸入元素的分區(qū)，其名稱為Partition。工作時除了提供一個StreamTransformation作為輸入外，還需要提供一個StreamPartitionor的實例來進行分區(qū)。

UnionTransformation

合并轉換器，該轉換器用于將多個輸入StreamTransformation進行合并，因此該轉換器接收StreamTransformation的集合。Union要求上游輸入的數據的結構必須是完全相同的。

FeedbackTransformation

表示FlinkDAG中的一個反饋點。簡單來說，就是把符合條件的數據發(fā)回上游Transformation處理，一個反饋點可以連接一個或多個上游的Transformation,這些連接關系叫反饋邊。符合反饋條件并交給上游的Transformation的數據流叫做反饋流。

FeedbackTransformation的固定名稱為Feedback

有兩個重要參數：

• input：上游輸入StreamTransformation
• waitTime：默認為0，即永遠等待，如果設置了等待時間，一旦超過該等待時間，則計算結束并且不再接收數據。

實例化FeedbackTransformation時，會自動創(chuàng)建一個用于存儲反饋邊的集合feedbackEdges。

FeedbackTransformation通過定義一個實力方法addFeedbackEdge來收集，

在加入的StreamTransformation的實例有一個要求，

當前FeedbackTransformation跟待加入的StreamTransformation并行度一致。

CoFeedbackTransformation

與FeedbackTransformation類似，也是FlinkDAG中的一個反饋點。

• 不同之處在于，CoFeedbackTransformation反饋給上游的數據流與上游Transformation的輸入類型不同
• 所以要求上游的Transformation必須是TwoInputTransformation。

四、算子行為

 4.1 生命周期管理

1）setup：初始化環(huán)境、時間服務、注冊監(jiān)控等。

2）open：該行為由各個具體的算子負責實現，包含了算子的初始化邏輯。

3）close：所有的數據處理完畢之后關閉算子，此時需要去報將所有的緩存數據向下游發(fā)送。

4）dispose：該方法在算子生命周期的最后執(zhí)行階段，此時算子已經關閉，停止處理數據，進行資源的釋放。

StreamTask作為算子的容器，負責管理算子的生命周期。

4.2 異步算子

異步算子的目的是解決與外部系統(tǒng)交互時網絡延遲所導致的系統(tǒng)瓶頸問題。

異步算子的兩種輸出模式

1）順序輸出

先收到的數據先輸出，后續(xù)數據元素的異步函數調用無論是否先完成，都需要等待，順序模式可以保證消息不亂序，但是可能增加延遲，降低算子的吞吐量。

2）無序輸出

先處理完的數據元素先輸出，不保證消息順序，相比于順序模式，無序輸出模式算子延遲低、吞吐量高。無序輸出模式并不是完全無序的，仍然要保持Watermark不能超越其前面數據元素的原則。等待完成隊列將按照Watermakr切分成組，組內可以無序輸出，組之間必須嚴格保證順序。

五、處理函數

5.1 雙流Join 即時Join

邏輯如下:

1) 創(chuàng)建一個State對象

2）接收到輸入流 1事件后更新Sate。

3）接收到輸出流 2的事件后遍歷State，根據Join條件進行匹配，將匹配結果發(fā)送到下游。

5.2延遲雙流Join

在流式數據里，數據可能是亂序的，數據會延遲到達，并且為了提供處理效率，使用小批量模式計算，而不是每個事件觸發(fā)一次Join計算。

邏輯如下:

1）創(chuàng)建2個state對象，分別緩存輸入流1和輸入流2的事件。

2）創(chuàng)建一個定時器，等待數據的到達，定時延遲觸發(fā)Join計算。

3）接收到輸入流1事件后更新State。

4）接收到輸入流2事件后更新State。

5）定時器遍歷State1和State2，根據Join條件進行匹配，將匹配結果發(fā)送到下游。

六、數據分區(qū)

數據分區(qū)在Flink中叫做Partition。

本質上說，分布式計算就是把一個作業(yè)切分成子任務Task，將不同的數據交給不同的Task計算。

StreamParitioner

是Flink中的數據流分區(qū)抽象接口，決定了在實際運行中的數據流分發(fā)模式。

自定義分區(qū)

使用用戶自定義分區(qū)函數，為每一個元組選擇目標分區(qū)。

ForwardParitioner

用于在同一個OperatorChain中上下游算子之間的數據轉發(fā)， 實際上數據是直接傳遞給下游的。

ShufflePartitioner

隨機將元素進行分區(qū)，可以確保下游的Task能夠均勻的獲取數據。

ReblancePartitioner

以Round-robin的方式為每個元素分配分區(qū)，確保下游的Task可以均勻的獲取數據，以免數據傾斜。

RescalingPartitioner

根據上下游Task的數據進行分區(qū)。

使用Round-robin選擇下游的一個Task進行數據分區(qū)，

如上游有2個Source,下游有6個Map,那么每個Source會分配3個固定下游的map,

不會向未分配給自己的分區(qū)寫入數據。

BroadcastPartitioner

將該記錄廣播給所有分區(qū)，即有N個分區(qū)，就把數據復制N份，每個分區(qū)1分

KeyGroupStreamPartitioner

keyedStream根據KeyGroup索引編號進行分區(qū)，該分區(qū)器不是提供給用戶來用。

KeyedStream在構造Transformation的時候默認使用KeyedGroup分區(qū)形式，從而在底層上支持作業(yè)Rescale功能。

七、分布式ID

到此這篇關于解析Flink內核原理與實現核心抽象的文章就介紹到這了,更多相關Flink內核原理核心抽象內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

最新評論