Flink 運行時環境由兩種類型進程組成，JobManager和TaskManager

• JobManager，也稱為 master，用于協調分布式執行。負責調度任務，檢查點，失敗恢復等。
• TaskManager，也稱為 worker，用于執行數據流圖的任務（更準確地說，是計算子任務），并對數據流進行緩沖、交換。Flink 運行環境中至少包含一個任務管理器。

Flink作業流程

Flink集群啟動時，會啟動一個JobManager進程、至少一個TaskManager進程。

在Local模式下，會在同一個JVM內部啟動一個JobManager進程和TaskManager進程。當Flink程序提交后，會創建一個Client來進行預處理，并轉換為一個并行數據流，這是對應著一個Flink Job，從而可以被JobManager和TaskManager執行。

在實現上，Flink基于Actor實現了JobManager和TaskManager，所以JobManager與TaskManager之間的信息交換，都是通過事件的方式來進行處理。 如上圖所示，Flink系統主要包含如下3個主要的進程：

關于JobManager

JobManager是Flink系統的協調者，它負責接收Flink Job，調度組成Job的多個Task的執行。同時，JobManager還負責收集Job的狀態信息，并管理Flink集群中從節點TaskManager。

JobManager所負責的各項管理功能，它接收到并處理的事件主要包括：

• RegisterTaskManager

在Flink集群啟動的時候，TaskManager會向JobManager注冊，如果注冊成功，則JobManager會向TaskManager回復消息AcknowledgeRegistration。

• SubmitJob

Flink程序內部通過Client向JobManager提交Flink Job，其中在消息SubmitJob中以JobGraph形式描述了Job的基本信息。

• CancelJob

請求取消一個Flink Job的執行，CancelJob消息中包含了Job的ID，如果成功則返回消息CancellationSuccess，失敗則返回消息CancellationFailure。

• UpdateTaskExecutionState

TaskManager會向JobManager請求更新ExecutionGraph中的ExecutionVertex的狀態信息，更新成功則返回true。

• RequestNextInputSplit

運行在TaskManager上面的Task，請求獲取下一個要處理的輸入Split，成功則返回NextInputSplit。

• JobStatusChanged

ExecutionGraph向JobManager發送該消息，用來表示Flink Job的狀態發生的變化，例如：RUNNING、CANCELING、FINISHED等。

關于TaskManager

TaskManager也是一個Actor，它是實際負責執行計算的Worker，在其上執行Flink Job的一組Task。每個TaskManager負責管理其所在節點上的資源信息，如內存、磁盤、網絡，在啟動的時候將資源的狀態向JobManager匯報。

TaskManager端可以分成兩個階段：

• 注冊階段

TaskManager會向JobManager注冊，發送RegisterTaskManager消息，等待JobManager返回AcknowledgeRegistration，然后TaskManager就可以進行初始化過程。

• 可操作階段

該階段TaskManager可以接收并處理與Task有關的消息，如SubmitTask、CancelTask、FailTask。如果TaskManager無法連接到JobManager，這是TaskManager就失去了與JobManager的聯系，會自動進入“注冊階段”，只有完成注冊才能繼續處理Task相關的消息。

Client

當用戶提交一個Flink程序時，會首先創建一個Client，該Client首先會對用戶提交的Flink程序進行預處理，并提交到Flink集群中處理，所以Client需要從用戶提交的Flink程序配置中獲取JobManager的地址，并建立到JobManager的連接，將Flink Job提交給JobManager。

Client會將用戶提交的Flink程序組裝一個JobGraph， 并且是以JobGraph的形式提交的。一個JobGraph是一個Flink Dataflow，它由多個JobVertex組成的DAG。其中，一個JobGraph包含了一個Flink程序的如下信息：JobID、Job名稱、配置信息、一組JobVertex等。

若干概念

任務槽與資源

每個 worker（任務管理器）都是一個獨立的 JVM 進程，每個子任務就是運行在其中的獨立線程里。為了控制 worker 接收任務的數量，在 worker 中引入了任務槽的概念（每個 worker 中至少包含一個任務槽）。

每個任務槽代表任務管理器中一個特定的資源池子集。例如，如果任務管理器有3個槽，它會為每個槽分配 1/3 的內存。將資源池槽化可以讓子任務獲取指定容量的內存資源，而避免同其他作業中的子任務競爭。注意，這里沒有對 CPU 進行隔離；目前任務槽僅僅用于劃分任務的內存。

通過調整任務槽的數量，用戶可以設定子任務之間獨立運行的程度。如果任務管理器中只有一個槽，那么每個任務組都會在一個獨立的 JVM（例如 JVM 可以在一個獨立的容器中啟動）中運行。任務管理器中配置更多的槽就意味著會有更多的子任務共享同一個 JVM。在同一個 JVM 中的任務會共享 TCP 連接（通過多路復用的方式）和心跳信息，同時他們也會共享數據集和數據結構，這在某種程度上可以降低單任務的開銷。

默認情況下，Flink 會允許同一個作業的多個子任務共享一個槽，即便這些子任務來自不同的任務。這種情況下，有可能會出現某個槽中包含一個完整的作業流水的場景。這樣做主要有兩點好處：

• Flink 集群需要在作業中確保任務槽數量和程序并發量完全一致，而并不需要計算程序中任務（每個任務的并發量也許都不相同）的具體數量。
• 可以提高資源利用率。如果沒有任務槽共享機制，非密集型的 source/map() 子任務就會和密集型的 window 子任務一樣阻塞大量資源。如果有任務槽共享機制，在程序的并發量從 2 提高到 6 的情況下（舉個例子），就可以讓密集型子任務完全分散到任務管理器中，從而可以顯著提高槽的資源利用率。

Flink API 中包含一個資源組機制，可以避免不合理的任務槽共享。

依照以往的經驗來說，默認的任務槽數量應設置為 CPU 核心的數量。如果使用超線程技術，每個槽中甚至可以調度處理超過 2 個硬件線程。

后端存儲

通過鍵值對索引的數據結構保存在選定的后端存儲中。有的后端存儲將數據保存在內存中的哈希表中，而有的存儲會使用 RocksDB 來保存鍵值對。除了定義保存狀態的數據結構之外，后端存儲還實現了獲取鍵值對的特定時間點快照的功能，該功能可以將快照保存為檢查點的一部分。

保存點

使用數據流 API 的程序可以從指定的保存點恢復。保存點具備更新程序和 Flink 集群而不丟失任何狀態的功能。

保存點可以看作是一種手動觸發的檢查點，該檢查點可以獲取程序的快照并將其寫入后端存儲中。所以說保存點的功能依賴于一般的檢查點機制。程序執行時會定期在 worker 節點生成快照和檢查點。由于 Flink 的恢復機制只需要使用最新一個有效的檢查點，在新的檢查點生成后就可以安全移除其余舊的檢查點了。

保存點和定期檢查點在大部分情況下都很相似，區別只在于保存點是由用戶觸發的，并且在新的檢查點生成后不會自動過期失效。保存點可以通過命令行生成，也可以在調用REST API取消作業時產生。

總結

以上是生活随笔為你收集整理的flink 运行一段时间 内存溢出_Flink之运行时环境的全部內容，希望文章能夠幫你解決所遇到的問題。

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