柚子快報激活碼778899分享：大數(shù)據(jù) flink尚硅谷

http://yzkb.51969.com/

flink

1 flink基礎使用1.1 角色1.2 部署模式（抽象）1.2.1 會話模式1.2.2 單作業(yè)模式1.2.3 應用模式

1.3 運行模式（實際 誰來管理資源）1.3.1 Stand alone1.3.2 YARN運行模式（重點）

2. 運行時架構2.1 系統(tǒng)架構2.2 核心概念2.2.1 并行度2.2.2 算子鏈2.2.2 任務槽

3 DataStream api3.1 創(chuàng)建環(huán)境3.2 source 讀取數(shù)據(jù)源【源算子】3.3 轉換算子 transformmapfilterflatmap

3.4 聚合算子keybysum/min/max/minby/maxbyreducerichMapFunction

3.5 分區(qū)3.6 分流、合流3.7 輸出算子

4. 窗口4.1 窗口分配器4.2 計算邏輯4.2.1 增量聚合4.2.2 全窗口聚合4.2.3 二者結合

4.3 關于按照數(shù)量的滑動窗口辨析4.4 觸發(fā)器和移除器4.5 窗口的實現(xiàn)細節(jié)

5. 水位線 watermark(針對事件時間)5.1 關于窗口與水位線的正確理解5.2 水位線生成策略5.3 水位線的傳遞5.4 遲到數(shù)據(jù)處理5.5 window join5.6 interval join5.7 關于水位線生成-1ms

6 處理函數(shù) process算子6.1 定時器6.2 窗口處理函數(shù)

7 狀態(tài)管理8 容錯機制

參考

1 flink基礎使用

flink是一種分布式任務處理引擎

事件驅動：有數(shù)據(jù)來才進行處理，不來就不動。程序一旦開始就會一直運行

現(xiàn)在使用的是datastreamsource作為數(shù)據(jù)源 demo：

讀socket這種無界流：

發(fā)送數(shù)據(jù)端：

flink端綁定一個無界數(shù)據(jù)源：

1.1 角色

三種角色：客戶端、job manager、task manager

客戶端提交job給job manager

1.2 部署模式（抽象）

1.2.1 會話模式

提前啟動一個flink集群，一旦提交一個任務就交給這個集群做

1.2.2 單作業(yè)模式

不提前啟動集群，一個作業(yè)提交了，為這個作業(yè)單獨啟動一個集群【每個作業(yè)都是隔離的，業(yè)務場景首選】 這種方式在flink1.17之后已經快要被淘汰了

需要注意的是，F(xiàn)lik本身無法直接這樣運行，所以單作業(yè)模式一般需要借助一些資源管理框架來啟動集群，比如YARN、Kubernetes(K8S)

1.2.3 應用模式

與（2）相比，代碼在客戶端解析==>代碼在job manager解析，其余一致

1.3 運行模式（實際 誰來管理資源）

1.3.1 Stand alone

（1）通過會話模式部署：

（2）不支持單作業(yè)模式部署：

（3）通過應用模式部署

按照如下命令啟動job manager 查看，發(fā)現(xiàn)應用模式部署成功

創(chuàng)建task manager：

1.3.2 YARN運行模式（重點）

可以通過三種部署模式實現(xiàn)YARN部署，區(qū)別在于起flink的指令參數(shù)不同

推薦應用模式部署

跟hdfs聯(lián)合使用，先把文件傳輸?shù)絟dfs

2. 運行時架構

2.1 系統(tǒng)架構

客戶端提交的一個job對應一個job master，job manager相當于進程，job master相當于線程 job manager中三大組件：分發(fā)器、job master、資源管理器 請求slot相當于請求資源 節(jié)點之間通過actor通訊系統(tǒng)進行溝通

2.2 核心概念

2.2.1 并行度

一個算子：指一種操作，比如sum/map/apply，每個操作都算是一個算子

keyby不算算子，只是一種轉換操作

一個算子包含多個子任務：指一個算子的活讓多個人去干，n個子任務干，并行度就叫n

一個流程的并行度是流程中每個步驟的并行度取max

默認并行度是cpu核數(shù)的二倍

2.2.2 算子鏈

上圖中，每一列是一個算子，每個圓圈是一個任務

前置知識-算子之間的兩種傳輸關系：

數(shù)據(jù)重分區(qū)可能是由并行度引起的，也可能是由keyby分組引起的（keyby導致數(shù)據(jù)重分區(qū)）

引入算子鏈： 并行度相同+一對一 針對某個算子前后禁用算子鏈：

從當前算子開始新鏈：

實際應用：兩個重量級算子不應該串在一起，斷開算子鏈還能幫助定位錯誤

2.2.2 任務槽

注意：slot大小是固定的，能夠均分隔離內存，但是不能隔離cpu

一個 TaskManager 可以同時執(zhí)行多個任務的子任務，這些子任務可以是不同的算子或操作符

不同算子的子任務才能處于同一個槽中，同一個槽里的三個任務同時進行，處理數(shù)據(jù)批次的進度不同

總結：

3 DataStream api

3.1 創(chuàng)建環(huán)境

StreamExecutionEnvironment.getExecutionEnvironment(conf) 自動分析是遠程集群還是本地idea環(huán)境 最后觸發(fā)執(zhí)行：env.execute()執(zhí)行當前flink job 如果想在一個main中執(zhí)行多個job【不建議】：env.executeAsync() 異步執(zhí)行獲取結果

流處理：一直來一直處理；批處理：一批到全了開始處理

3.2 source 讀取數(shù)據(jù)源【源算子】

// 從集合讀

DataStreamSource sc = env.fromCollection(array);

//從文件讀 依賴于FileSource

DataStreamSource sc = env.fromSource(source的實現(xiàn)類(FileSource build出來的),watermark,名字);

//從socket讀數(shù)據(jù)

DataStream sc = env.socketTextStream("localhost", 7777);

從Kafka讀：第三方件

3.3 轉換算子 transform

map

一進一出

實現(xiàn)mapfunction接口

filter

對每個元素進行判斷，true保留反之丟棄

實現(xiàn)filterfunction接口

flatmap

扁平映射，一進多出

實現(xiàn)flatmapfunction接口

辨析：與map區(qū)別： map使用return控制一進一出 flatmap沒有返回值 使用collector來收集輸出

3.4 聚合算子

keyby

經過keyby才有后面的聚合算子，保證相同key發(fā)往同一個分區(qū)（即同一個子任務），但是不同key也可能在同一個分區(qū)

實現(xiàn)keySelector接口

keyby不是轉換算子，因此不能設置并行度，只是對數(shù)據(jù)進行重分區(qū)

keyby之后得到keyedStream

sum/min/max/minby/maxby

keyby之后調用的

max：取比較字段最大值，非比較字段取第一次的值 maxby：取比較字段最大值，非比較字段取最大值字段所在數(shù)據(jù)行的數(shù)值

reduce

keyby之后調用的，輸入輸出類型需要保持一致，內部聚合邏輯比較靈活

reduce方法用于所有數(shù)據(jù)兩兩結合，前面累計的結果作為reduce函數(shù)的第一個參數(shù)，現(xiàn)在來的數(shù)據(jù)作為第二個參數(shù)

實現(xiàn)reduce Function接口

richMapFunction

為什么rich？多了生命周期管理方法，提供了子任務開啟與關閉前的操作函數(shù)open、close，多了運行時上下文，獲取運行時的環(huán)境信息

3.5 分區(qū)

分區(qū)算子 shuffle/rebalance（數(shù)據(jù)傾斜時使下游消費數(shù)據(jù)時是均勻的）/rescale（局部組隊）

自定義分區(qū)器，自己實現(xiàn)分區(qū)邏輯：

簡化后：

3.6 分流、合流

分流：將數(shù)字分為奇數(shù)流和偶數(shù)流

沒有合適算子，就用最底層的process算子

數(shù)據(jù)分流邏輯需要依賴支流的tag，這樣才能標記往哪里放：

辨析：filter與分流方式的區(qū)別：前者幾條流就要處理幾遍，分流只需處理數(shù)據(jù)一遍

數(shù)據(jù)類型相同才能合流：

union合并多個流 connect合并兩個流（工作場景常用）

可編輯程度更高的demo，多并行度的情況下必須結合keyby使用：

如果多并行度下沒有結合keyby，會導致應該關聯(lián)在一起的記錄沒有分在同一個分區(qū)（子任務）：

添加keyby操作：

3.7 輸出算子

輸出到文件：

紅框為必須設置，文件滾動策略表示達到標準后文件就結束

輸出到Kafka：

輸出到MySQL：

自定義sink邏輯：

RichSinkFunction帶來的open與close方法是子任務開啟/關閉時調用的

4. 窗口

分類：時間窗口，計數(shù)窗口 滾動窗口（固定大小，窗口無重疊）、滑動窗口（可以重疊，設置滑動步長）、會話窗口（只能基于時間，指定時間內沒有數(shù)據(jù)來，則當前窗口關閉）

4.1 窗口分配器

這里進行窗口的定義

經過keyby的窗口：每個key單開一個窗口 沒有keyby的窗口，所有數(shù)據(jù)不分流，公用一個窗口：

窗口劃分后的stream類型轉換：

4.2 計算邏輯

4.2.1 增量聚合

reduce（強制輸入類型、中間變量類型和輸出類型全都保持一致）：

demo：滾動窗口，5s一個

aggregate：（比reduce更靈活）

4.2.2 全窗口聚合

窗口內統(tǒng)一計算，可以獲取上下文context

4.2.3 二者結合

既有增量（來一個處理一個節(jié)省空間）又有全量（上下文）

第一個參數(shù)是aggregateFunction（接口），第二個參數(shù)是processWindowFunction（接口）

4.3 關于按照數(shù)量的滑動窗口辨析

以長度為5步長為2的計數(shù)窗口舉例：

每經過一個步長，都有一個窗口觸發(fā)輸出，第一次輸出在第二條數(shù)據(jù)來的時候（想象前面有三個空數(shù)據(jù)）

第一條數(shù)據(jù)不代表是第一個窗口的起始，12兩條數(shù)據(jù)是第一個窗口的結尾

4.4 觸發(fā)器和移除器

觸發(fā)器：

4.5 窗口的實現(xiàn)細節(jié)

5. 水位線 watermark(針對事件時間)

兩種時間語義：事件時間&處理時間

水位線是插入到數(shù)據(jù)流中的一個標記，可以認為是特殊的數(shù)據(jù)。主要內容就是一個時間戳，用于表示當前事件時間的進展（必須單調遞增）

watermark計算方式：當前最大事件時間 - 等待時間

參考

水位線的制作：

理想狀態(tài)，數(shù)據(jù)量小，來一條生產一條水位線

數(shù)據(jù)量大，隔一段時間生成一個水位線

亂序流：

亂序+數(shù)據(jù)量?。?/p>

亂序+數(shù)據(jù)量大：

取當前數(shù)據(jù)中最大時間戳

亂序流+允許遲到：

eg. 窗口長度為10，允許遲到兩秒，當?shù)?0秒生成的數(shù)據(jù)到來時并不會觸發(fā)窗口的關閉，因為等待機制會將當前時間視為10-2=8 當?shù)?2秒生產的數(shù)據(jù)來的時候才會觸發(fā)窗口關閉，因為等待機制下當前時間為12-2=10

5.1 關于窗口與水位線的正確理解

參考

某一時刻可能存在多個窗口桶，水位線是用于控制窗口的關閉的

5.2 水位線生成策略

有序流demo：

亂序流demo：有等待時間

也可以在數(shù)據(jù)源階段就指定水位線

5.3 水位線的傳遞

當一個子任務上游有多個子任務（水位線各不相同，即數(shù)據(jù)處理進度各不相同），取上游子任務最小的水位線作為當前子任務的水位線，再廣播給所有需要的下游

超過等待時間，則不把這個上游子任務的水位線作為當前水位線的min候選集

forMonotonousTimestamps() 只能有序數(shù)據(jù)，水位線也是升序的，沒有等待時間 forBoundedOutOfOrderness(3) 允許亂序數(shù)據(jù)，設置等待時間

5.4 遲到數(shù)據(jù)處理

允許窗口晚關窗，晚到的數(shù)據(jù)額外處理，關窗后再來的數(shù)據(jù)不再處理：

辨析：window晚關窗和watermark等待有什么區(qū)別？

側輸出流（基于tag）：

5.5 window join

合并兩個流，將兩條流的對應窗口內的內容兩兩合并

5.6 interval join

只支持事件時間語義

遲到數(shù)據(jù)處理邏輯：

5.7 關于水位線生成-1ms

亂序流中生成的水位線的時間戳，其實是當前數(shù)據(jù)中 最大時間戳 – 延遲時間 – 1ms

6 處理函數(shù) process算子

是最底層算子，自由度最高

6.1 定時器

只有keyed之后的數(shù)據(jù)流才能定義定時器

注冊定時器： 時間達到閾值后會觸發(fā)指定事件。這個時間閾值可以是事件時間也可以是處理時間。同key定時器會去重。 定義具體事件

6.2 窗口處理函數(shù)

以topn問題為例：

方法一： 定義10s的事件滑動窗口（可重疊），滑動步長為5s+windowall全窗口函數(shù)（不能key分區(qū)） 自己實現(xiàn)處理所有窗口的接口 方法二：使用keyedProcessFunction實現(xiàn)

增量計算+全量打標簽

增量：來一個數(shù)據(jù)計算一次，三種類型可不同（輸入類型，中間類型，輸出類型）

AggregateFunction 是用于執(zhí)行聚合操作的接口。它允許您定義如何將一個輸入值序列聚合為一個輸出值。

全窗口函數(shù)：處理窗口內的統(tǒng)計性結果

ProcessWindowFunction 會接收窗口中的所有元素，并允許您在窗口關閉時執(zhí)行自定義的處理邏輯。 調用以上兩個方法：

定義window內部處理函數(shù)：

process里放下面這個topn方法

KeyedProcessFunction 是窗口無關的，它是用于處理基于鍵控流的自定義處理函數(shù)

方法二總結：

7 狀態(tài)管理

8 容錯機制

柚子快報激活碼778899分享：大數(shù)據(jù) flink尚硅谷

http://yzkb.51969.com/

參考閱讀