Flink API提供了開發的介面，此外，為了實作業務邏輯，還必須為開發者提供自定義業務邏輯的能力，，Flink中設計了用戶自定義函式體系(User Defined Function,UDF),開發人員實作業務邏輯就是開發UDF，

一、環境物件

StreamExecutionEnvironment是Flink應用開發時的概念，表示流計算作業的執行環境，是作業開發的入口、資料源介面、生成和轉換DataStream的介面、資料Sink的介面、作業配置介面、作業啟動執行的入口，

Environment是運行時作業級別的概念，從StreamExecutionEnvironment中的配置資訊衍生而來，進入到Flink作業執行的時刻，作業需要的是相關的配置資訊，如作業的名稱、并行度、作業編號JobID、監控的Metric、容錯的配置資訊、IO等，用StreamExecutionRuntime物件就不適合了，很多API是不需要的，所以在Flink中抽象出了Environment作為運行時刻的背景關系資訊，

RuntimeContext是運行時Task實體級別的概念，Environment本身仍然是比較粗粒度作業級別的配置，對于每一個Task而言，其本身有更細節的配置資訊,所以Flink又抽象了RuntimeContext,每一個Task實體有自己的RuntimeContext，

環境物件關系如下：

1.1 執行環境

StreamExecutionEnvironment是Flink流計算應用的執行環境，是Flink作業開發和啟動執行的入口，開發者對StreamExecutionEnvironment的實作是無感知的，

• LocalStreamEnvironment

本地執行環境，在單個JVM中使用多執行緒模擬Flink集群，

其基本的作業流程如下：

1) 執行Flink作業的Main函式生成Streamgraph,轉化為JobGraph，

2) 設定任務運行的配置資訊，

3) 根據配置資訊啟動對應的LocalFlinkMiniCluster，

4) 根據配置資訊和miniCluster生成對應的MiniClusterClient，

5) 通過MiniClusterClient提交JobGraph 到MiniCluster，

• RemoteStreamEnvironment

在大規模資料中心中部署的Flink生產集群的執行環境，

當將作業發布到Flink集群的時候，使用RemoteStreamEnvironment，

其基本的作業流程如下：

1) 執行Flink作業的Main函式生成Streamgraph，轉化為JobGraph，

2) 設定任務運行的配置資訊，

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

• StreamContextEnvironment

在Cli命令列或者單元測驗時候會被使用，執行步驟同上，

• StreamPlanEnvironment

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

• ScalaShellStreamEnvironment

這是Scala Shell執行環境，可以在命令列中互動式開發Flink作業，

其基本作業流程如下：

1) 校驗部署模式，目前Scala Shell僅支持attached模式，

2) 上傳每個作業需要的Jar檔案，

其余步驟與RemoteStreamEnvironment類似，

1.2 運行時環境

• RuntimeEnvironment

在Task開始執行時進行初始化，把Task運行相關的資訊都封裝到該物件中，其中不光包含了配置資訊，運行時的各種服務也會被包裝到其中，

• SavepointEnvironment

SavepointEnvironment是Environment的最小化實作，在狀態處理器的API中使用，

1.3 運行時背景關系

RuntimeContext是Function運行時的背景關系，封裝了Function運行時可能需要的所有資訊，讓Function在運行時能夠獲取到作業級別的資訊，如并行度相關資訊、Task名稱、執行配置資訊(ExecutionConfig)、State等，

Function的每個實體都有一個RuntimeContext物件，在RichFunction中通過getRunctionContext()可以訪問該物件，

RuntimeContext的類體系圖如下:

• StreamingRuntimeContext：在流計算UDF中使用的背景關系，用來訪問作業資訊、狀態等，
• DistributedRuntimeUDFContext：由運行時UDF所在的批處理算子創建，在DataSet批處理中使用，
• RuntimeUDFContext：在批處理應用的UDF中使用，
• SavepointRuntimeContext：支持對檢查點和保存點進行操作，包括讀取、變更、寫入等，
• CepRuntimeContext：CEP復雜事件處理中使用的背景關系，

二、資料流元素

資料流元素在Flink中叫做StreamElement，有資料記錄StreamRecord,延遲標記LatencyMarker、Watermark、流狀態標記StreamStatus這四種，在執行層面，4種資料流元素都被序列化成二進制資料，形成混合的資料流，在算子中將混合資料流中的資料流元素反序列化出來，

• StreamRecord

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

包含以下內容：

1）資料的值本身

2）時間戳(可選)

• LatencyMarker

用來近似評估延遲，LatencyMarker在Source中創建，并向下游發送，繞過業務處理邏輯，在Sink節點中使用LatencyMarker估計資料在整個DAG圖中的流轉花費的時間，

LatencyMarker包含資訊如下：

1）周期性的在資料源算子中創造出來的時間戳，

2）算子編號

3）資料源算子所在的Task編號

• Watermark

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

• StreamStatus

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

有兩種表示狀態:

1）空閑狀態(IDLE)

2）活動狀態(ACTIVE)

三、資料轉換

資料轉換在Flink中叫做Transformation,是銜接DataStream Api和Flink內核的邏輯結構，

Transformation有兩大類：

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

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

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

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

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

3）bufferTimeout：buffer超時時間，

4）parallelism：并行度，

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

6）outputType：輸出型別，用來進行序列化資料，

7）slotSharingGroup：給當前的Transformation設定Slot共享組，

3.1 物理Transformation

• SourceTransformation

從資料源讀取資料的Transformation，是Flink作業的起點，只有下游Transformation，沒有上游輸入，

• SinkTransformation

將資料寫到外部存盤的Transformation,是Flink作業的終點，

• OneInputTransformation

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

• TwoInputTransformation

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

3.2 虛擬Transformation

• SideOutputTransformation

在旁路輸出中轉換而來，表示上游Transformation的一個分流，每個sideoutput通過OutputTag標識，

• SplitTransformation

用來按條件切分資料流，該轉換用于將一個流拆分成多個流，

• SelectTransformation

與SplitTransformation配合使用，用來在下游選擇SplitTransformation切分的資料流，

• PartitionTransformation

該轉換器用于改變輸入元素的磁區，其名稱為Partition，作業時除了提供一個StreamTransformation作為輸入外，還需要提供一個StreamPartitionor的實體來進行磁區，

• UnionTransformation

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

• FeedbackTransformation

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

FeedbackTransformation的固定名稱為Feedback，有兩個重要引數：

1）input：上游輸入StreamTransformation

2）waitTime：默認為0，即永遠等待，如果設定了等待時間，一旦超過該等待時間，則計算結束并且不再接收資料，

實體化FeedbackTransformation時，會自動創建一個用于存盤反饋邊的集合feedbackEdges，FeedbackTransformation通過定義一個實力方法addFeedbackEdge來收集，在加入的StreamTransformation的實體有一個要求，當前FeedbackTransformation跟待加入的StreamTransformation并行度一致，

• CoFeedbackTransformation

與FeedbackTransformation類似，也是FlinkDAG中的一個反饋點，不同之處在于，CoFeedbackTransformation反饋給上游的資料流與上游Transformation的輸入型別不同，所以要求上游的Transformation必須是TwoInputTransformation，

四、算子行為

4.1 生命周期管理

1）setup：初始化環境、時間服務、注冊監控等，

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

3）close：所有的資料處理完畢之后關閉算子，此時需要去報將所有的快取資料向下游發送，

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

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

4.2 異步算子

異步算子的目的是解決與外部系統互動時網路延遲所導致的系統瓶頸問題，

異步算子的兩種輸出模式

1）順序輸出

先收到的資料先輸出，后續資料元素的異步函式呼叫無論是否先完成，都需要等待，順序模式可以保證訊息不亂序，但是可能增加延遲，降低算子的吞吐量，

2）無序輸出

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

五、處理函式

5.1 雙流Join

• 即時Join

邏輯如下:

1) 創建一個State物件

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

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

• 延遲雙流Join

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

邏輯如下:

1）創建2個state物件，分別快取輸入流1和輸入流2的事件，

2）創建一個定時器，等待資料的到達，定時延遲觸發Join計算，

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

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

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

六、資料磁區

資料磁區在Flink中叫做Partition，本質上說，分布式計算就是把一個作業切分成子任務Task，將不同的資料交給不同的Task計算，StreamParitioner是Flink中的資料流磁區抽象介面，決定了在實際運行中的資料流分發模式，

• 自定義磁區

使用用戶自定義磁區函式，為每一個元組選擇目標磁區，

• ForwardParitioner

用于在同一個OperatorChain中上下游算子之間的資料轉發， 實際上資料是直接傳遞給下游的，

• ShufflePartitioner

隨機將元素進行磁區，可以確保下游的Task能夠均勻的獲取資料，

• ReblancePartitioner

以Round-robin的方式為每個元素分配磁區，確保下游的Task可以均勻的獲取資料，以免資料傾斜，

• RescalingPartitioner

根據上下游Task的資料進行磁區，使用Round-robin選擇下游的一個Task進行資料磁區，如上游有2個Source,下游有6個Map,那么每個Source會分配3個固定下游的map,不會向未分配給自己的磁區寫入資料，

• BroadcastPartitioner

將該記錄廣播給所有磁區，即有N個磁區，就把資料復制N份，每個磁區1分

• KeyGroupStreamPartitioner

keyedStream根據KeyGroup索引編號進行磁區，該磁區器不是提供給用戶來用，KeyedStream在構造Transformation的時候默認使用KeyedGroup磁區形式，從而在底層上支持作業Rescale功能，

七、分布式ID

接下來Flink時間視窗篇，如果對Flink感興趣或者正在使用的小伙伴，可以加我入群一起探討學習，

參考書籍《Flink 內核原理與實作》