Flink并行度優先級(從高到低)

sum(1).setParallelism(1) env.setParallelism(1) ApacheFlinkDashboard任務添加并行度配置 flink-conf.yaml并行度配置 注: 處理輸入輸出時, 并行度默認為 1

Flink集群常用指令

提交任務 run: 代表執行; c: 指定入口類; p: 并行度; host, post: 主機地址端口 flink run -c com.test.StreamWordCount -p 1 /home/Project/Flink/target/FlinkStudy-1.0-SNASHOT.jar --host localhost --port 7777取消任務 flink cancel jobId列出所有 jobId flink list

Flink部署

Standalone 模式: TaskManager固定, 所有任務共享 Dispatcher 和 ResourceManager, 當任務滿了, 剩余任務只能阻塞等待.

Yarn 模式:
a. Session 模式: 與Standalone模式一樣, 不同點在于 TaskManager 進行動態分配, 提高集群擴展能力.
b. Job 模式: 每個提交的任務單獨有一份 Dispatcher, ResourceManager, 適用于特別大的任務, 且每個任務執行耗時很長. 注: job模式下, 不用啟動 yarn-session, 直接 flink run -m yarn-cluster -c提交.

Kubernetes 模式
此處具體部署方式, 未貼出, 詳情可見官網說明.

Flink運行組件

JobManager: 作業管理器, 負責作業管理
控制提交的 job 執行, JobManager 向 ResourceManager 請求 TaskManager 上的 slot, 獲取到足夠資源, 將該任務的執行圖分發到運行的 TaskManager 上, 運行中, JobManager 負責所有需要中央協調的操作(例: checkpoints 協調, 存盤, 故障檢測)

TaskManager: 任務管理器, 負責干活
Flink 中會有多個 TaskManager 執行, 且每個 TaskManager 都包含一定數量的 slot, slot 限制了 TaskManager 能夠執行的任務數量.
TaskManager啟動后, 向 ResourceManager 注冊 slot, 根據 ResourceManager 調配, 將 slot 提供給 JobManager 調用, JobManager 想 slot 分配任務執行.
任務在各個執行階段執行可能使用不同的 slot 執行, 在同一任務下, 不同的 TaskManager 可進行數據交換.

ResourceManager: 資源管理器, 分發資源, 調配 slot
管理 TaskManager 的 slot.
在不同的環境下(Yarn, K8s, Standalone), 提供不同的 ResourceManager.
JobManager 申請資源, ResourceManager 將空閑的 slot 分配給 JobManager, 若 slot 不足, 向資源提供平臺發起會話, 提供滿足的容器

Dispatcher: 分發器
為應用提交提供一個 REST 接口, 也會啟動一個 webUI 便于展示提交的任務, 該組件在架構中可以是非必需的. 任務被提交時, 將給任務移交給 JobManager

任務提交流程

抽象模式的任務提交流程

Yarn模型下的任務提交

任務調度原理

任務調度流程:

并行度(Parallelism): 執行算子的子任務個數; 所有算子中最大并行度就是整個 Stream 并行度.

TaskManager, Slots的任務調度: Flink 允許子任務共享 slot(不同任務的子任務也能共享), 由于可以共享, 一個 slot 可保存整個作業流程.
子任務不共享, 如下圖:

子任務共享 slot, 一個 slot 保存一個工作流程, 如下圖, 這種情況可極大提高任務并行度

代碼設置不同子任務共享一個 slot, 代碼如下, 此時 有 2 個 slot, default 和 A, default 處理輸入流, A 處理 flatMap 和 sum

//輸入流未設置 slot 組, 使用 default. DataStreamSource<String> source = env.socketTextStream("localhost", 7777);DataStream<Tuple2<String, Integer>> resultStream = source.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {public void flatMap(String value, Col lector<Tuple2<String, Integer>> out) throws Exception {String[] words = value.split("\\W+");for (String s : words) {out.collect(new Tuple2<String, Integer>(s, 1));}}// flatMap 與 sum 求和共享一個 slot}).slotSharingGroup("A").keyBy(0).sum(1).slotSharingGroup("A");

數據流(DataFlow)

所有的 Flink 程序都由 3 部分組成: Source(數據讀取), Transformation(數據加工), Sink(輸出)

Flink 上運行程序都會被映射為 DataFlow, 一個 DataFlow 由一個或多個 source 開始, sink 結束.

大部分情況, 程序中的 transformation 與 DataFlow 的 operator 一一對應.

數據流圖變化過程
StreamGraph -> JobGraph -> ExecutionGraph -> 物理執行圖
StreamGraph: 根據 Stream API 生成的程序拓撲結構.
JobGraph: StreamGraph 優化得到(優化措施: 將多個符合條件的節點合并為一個節點), 提交給 JobManager 的數據結構.
ExecutionGraph: JobManager 根據 JobGraph 生成 ExecutionGraph(JobGraph 的并行版本)
物理執行圖: JobManager 根據 ExecutionGraph 對 Job 進行調度, 在各個 TaskManager 上部署 Task 后形成的, 非具體數據結構.
例如: 單詞計數數據流執行圖變化過程

算子合并

數據傳輸形式

一個程序中, 不同算子存在不同并行度

算子之間傳輸數據模式: one-to-one, redistributing
one-to-one: 不會打亂元素, 分區順序, 例如: map, filter, flatMap
redistributing: stream 分區會改變(數據下游存在多個并行分區, 輪詢/隨機方式往下游發送數據), 例如: keyBy 基于 hashCode 重分區, broadcast, rebalance 會隨機重新分區.

任務鏈

Flink 使用任務鏈將多個相同并行度的 one-to-one 操作合并, 使其構成一個 task, 之前的算子操作變為 subTask

并行度相同, one-to-one 操作, 且在同一個 slot 共享組, 三者是必須的.

注: 如果不想算子合并, 可調整算子間并行度, 設置重分區, 或在算子間添加 disableChaining()方法

總結

以上是生活随笔為你收集整理的Flink并行度优先级_集群操作常用指令_运行组件_任务提交流程_数据流图变化过程的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。