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

一、環(huán)境對象

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

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

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

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

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

StreamExecutionEnvironment

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

開發(fā)者對StreamExecutionEnvironment的實(shí)現(xiàn)是無感知的。

LocalStreamEnvironment

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

其基本的工作流程如下：

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

2) 設(shè)置任務(wù)運(yùn)行的配置信息。

3) 根據(jù)配置信息啟動對應(yīng)的LocalFlinkMiniCluster。

4) 根據(jù)配置信息和miniCluster生成對應(yīng)的MiniClusterClient。

5) 通過MiniClusterClient提交JobGraph 到MiniCluster。

RemoteStreamEnvironment

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

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

其基本的工作流程如下：

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

2) 設(shè)置任務(wù)運(yùn)行的配置信息。

3) 提交JobGraph到遠(yuǎn)程的Flink集群。

StreamContextEnvironment

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

StreamPlanEnvironment

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

ScalaShellStreamEnvironment

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

其基本工作流程如下：

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

其余步驟與RemoteStreamEnvironment類似。

1.2 運(yùn)行時(shí)環(huán)境

RuntimeEnvironment

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

SavepointEnvironment

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

1.3 運(yùn)行時(shí)上下文

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

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

RuntimeContext的類體系圖如下:

StreamingRuntimeContext：

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

DistributedRuntimeUDFContext：

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

RuntimeUDFContext：                                                          

在批處理應(yīng)用的UDF中使用。

SavepointRuntimeContext：

支持對檢查點(diǎn)和保存點(diǎn)進(jìn)行操作，包括讀取、變更、寫入等。

CepRuntimeContext：

CEP復(fù)雜事件處理中使用的上下文。

二、數(shù)據(jù)流元素

數(shù)據(jù)流元素在Flink中叫做StreamElement

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

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

StreamRecord

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

包含以下內(nèi)容：

1）數(shù)據(jù)的值本身

2）時(shí)間戳(可選)

LatencyMarker

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

LatencyMarker包含信息如下：

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

2）算子編號

3）數(shù)據(jù)源算子所在的Task編號

Watermark

是一個(gè)時(shí)間戳，用來告訴算子所有時(shí)間早于等于Watermark的事件或記錄都已經(jīng)達(dá)到，不會再有比Watermark更早的記錄。

StreamStatus

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

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

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

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

三、數(shù)據(jù)轉(zhuǎn)換

數(shù)據(jù)轉(zhuǎn)換在Flink中叫做Transformation,是銜接DataStream Api和Flink內(nèi)核的邏輯結(jié)構(gòu)。

Transformation有兩大類：

1）物理Transformation: 會轉(zhuǎn)換成算子，繼承了PhysicalTransformation。

2）虛擬Transformation: 不會轉(zhuǎn)換成具體算子。

Tranformation包含了Flink的運(yùn)行時(shí)關(guān)鍵參數(shù):

1）name：轉(zhuǎn)換器名稱，主要用于可視化。

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

3）bufferTimeout：buffer超時(shí)時(shí)間。

4）parallelism：并行度。

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

6）outputType：輸出類型，用來進(jìn)行序列化數(shù)據(jù)。

7）slotSharingGroup：給當(dāng)前的Transformation設(shè)置Slot共享組。

3.1 物理Transformation SourceTransformation

從數(shù)據(jù)源讀取數(shù)據(jù)的Transformation，是Flink作業(yè)的起點(diǎn)。

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

SinkTransformation

將數(shù)據(jù)寫到外部存儲的Transformation,是Flink作業(yè)的終點(diǎn)。

OneInputTransformation

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

TwoInputTransformation

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

3.2 虛擬Transformation SideOutputTransformation

在旁路輸出中轉(zhuǎn)換而來，表示上游Transformation的一個(gè)分流。

每個(gè)sideoutput通過OutputTag標(biāo)識。

SplitTransformation

用來按條件切分?jǐn)?shù)據(jù)流，該轉(zhuǎn)換用于將一個(gè)流拆分成多個(gè)流。

SelectTransformation

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

PartitionTransformation

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

UnionTransformation

合并轉(zhuǎn)換器，該轉(zhuǎn)換器用于將多個(gè)輸入StreamTransformation進(jìn)行合并，因此該轉(zhuǎn)換器接收StreamTransformation的集合。Union要求上游輸入的數(shù)據(jù)的結(jié)構(gòu)必須是完全相同的。

FeedbackTransformation

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

FeedbackTransformation的固定名稱為Feedback

有兩個(gè)重要參數(shù)：

• input：上游輸入StreamTransformation
• waitTime：默認(rèn)為0，即永遠(yuǎn)等待，如果設(shè)置了等待時(shí)間，一旦超過該等待時(shí)間，則計(jì)算結(jié)束并且不再接收數(shù)據(jù)。

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

FeedbackTransformation通過定義一個(gè)實(shí)力方法addFeedbackEdge來收集，

在加入的StreamTransformation的實(shí)例有一個(gè)要求，

當(dāng)前FeedbackTransformation跟待加入的StreamTransformation并行度一致。

CoFeedbackTransformation

與FeedbackTransformation類似，也是FlinkDAG中的一個(gè)反饋點(diǎn)。

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

四、算子行為

 4.1 生命周期管理

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

2）open：該行為由各個(gè)具體的算子負(fù)責(zé)實(shí)現(xiàn)，包含了算子的初始化邏輯。

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

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

StreamTask作為算子的容器，負(fù)責(zé)管理算子的生命周期。

4.2 異步算子

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

異步算子的兩種輸出模式

1）順序輸出

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

2）無序輸出

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

五、處理函數(shù)

5.1 雙流Join 即時(shí)Join

邏輯如下:

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

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

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

5.2延遲雙流Join

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

邏輯如下:

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

2）創(chuàng)建一個(gè)定時(shí)器，等待數(shù)據(jù)的到達(dá)，定時(shí)延遲觸發(fā)Join計(jì)算。

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

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

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

六、數(shù)據(jù)分區(qū)

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

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

StreamParitioner

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

自定義分區(qū)

使用用戶自定義分區(qū)函數(shù)，為每一個(gè)元組選擇目標(biāo)分區(qū)。

ForwardParitioner

用于在同一個(gè)OperatorChain中上下游算子之間的數(shù)據(jù)轉(zhuǎn)發(fā)， 實(shí)際上數(shù)據(jù)是直接傳遞給下游的。

ShufflePartitioner

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

ReblancePartitioner

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

RescalingPartitioner

根據(jù)上下游Task的數(shù)據(jù)進(jìn)行分區(qū)。

使用Round-robin選擇下游的一個(gè)Task進(jìn)行數(shù)據(jù)分區(qū)，

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

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

BroadcastPartitioner

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

KeyGroupStreamPartitioner

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

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

七、分布式ID

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

最新評論