這期內(nèi)容當(dāng)中小編將會(huì)給大家?guī)碛嘘P(guān)如何進(jìn)行Flink實(shí)時(shí)任務(wù)資源優(yōu)化探索與實(shí)踐，文章內(nèi)容豐富且以專業(yè)的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

新賓網(wǎng)站建設(shè)公司成都創(chuàng)新互聯(lián),新賓網(wǎng)站設(shè)計(jì)制作，有大型網(wǎng)站制作公司豐富經(jīng)驗(yàn)。已為新賓上千家提供企業(yè)網(wǎng)站建設(shè)服務(wù)。企業(yè)網(wǎng)站搭建\外貿(mào)網(wǎng)站建設(shè)要多少錢，請(qǐng)找那個(gè)售后服務(wù)好的新賓做網(wǎng)站的公司定做！

隨著 Flink K8s 化以及實(shí)時(shí)集群遷移完成，有贊越來越多的 Flink 實(shí)時(shí)任務(wù)運(yùn)行在 K8s 集群上，F(xiàn)link K8s 化提升了實(shí)時(shí)集群在大促時(shí)彈性擴(kuò)縮容能力，更好的降低大促期間機(jī)器擴(kuò)縮容的成本。同時(shí)，由于 K8s 在公司內(nèi)部有專門的團(tuán)隊(duì)進(jìn)行維護(hù)， Flink K8s 化也能夠更好的減低公司的運(yùn)維成本。

不過當(dāng)前 Flink K8s 任務(wù)資源是用戶在實(shí)時(shí)平臺(tái)端進(jìn)行配置，用戶本身對(duì)于實(shí)時(shí)任務(wù)具體配置多少資源經(jīng)驗(yàn)較少，所以存在用戶資源配置較多，但實(shí)際使用不到的情形。比如一個(gè) Flink 任務(wù)實(shí)際上 4 個(gè)并發(fā)能夠滿足業(yè)務(wù)處理需求，結(jié)果用戶配置了 16 個(gè)并發(fā)，這種情況會(huì)導(dǎo)致實(shí)時(shí)計(jì)算資源的浪費(fèi)，從而對(duì)于實(shí)時(shí)集群資源水位以及底層機(jī)器成本，都有一定影響。基于這樣的背景，小編從 Flink 任務(wù)內(nèi)存以及消息能力處理方面，對(duì) Flink 任務(wù)資源優(yōu)化進(jìn)行探索與實(shí)踐。

一、Flink 計(jì)算資源類型與優(yōu)化思路

1.1 Flink 計(jì)算資源類型

一個(gè) Flink 任務(wù)的運(yùn)行，所需要的資源我認(rèn)為能夠分為 5 類：

1. 內(nèi)存資源

2. 本地磁盤（或云盤）存儲(chǔ)

3. 依賴的外部存儲(chǔ)資源。比如 HDFS、S3 等（任務(wù)狀態(tài)/數(shù)據(jù)），HBase、MySQL、redis 等（數(shù)據(jù)）

4. CPU 資源

5. 網(wǎng)卡資源

目前 Flink 任務(wù)使用最主要的還是內(nèi)存和 CPU 資源，本地磁盤、依賴的外部存儲(chǔ)資源以及網(wǎng)卡資源一般都不會(huì)是瓶頸，所以本文我們是從 Flink 任務(wù)的內(nèi)存和 CPU 資源，兩個(gè)方面來對(duì) Flink 實(shí)時(shí)任務(wù)資源進(jìn)行優(yōu)化。

1.2 Flink 實(shí)時(shí)任務(wù)資源優(yōu)化思路

對(duì)于 Flink 實(shí)時(shí)任務(wù)資源分析思路，我們認(rèn)為主要包含兩點(diǎn)：

• 一是從任務(wù)內(nèi)存視角，從堆內(nèi)存方面對(duì)實(shí)時(shí)任務(wù)進(jìn)行分析。

• 另一方面則是從實(shí)時(shí)任務(wù)消息處理能力入手，保證滿足業(yè)務(wù)方數(shù)據(jù)處理需求的同時(shí)，盡可能合理使用 CPU 資源。

之后再結(jié)合實(shí)時(shí)任務(wù)內(nèi)存分析所得相關(guān)指標(biāo)、實(shí)時(shí)任務(wù)并發(fā)度的合理性，得出一個(gè)實(shí)時(shí)任務(wù)資源預(yù)設(shè)值，在和業(yè)務(wù)方充分溝通后，調(diào)整實(shí)時(shí)任務(wù)資源，最終達(dá)到實(shí)時(shí)任務(wù)資源配置合理化的目的，從而更好的降低機(jī)器使用成本。

1.2.1 任務(wù)內(nèi)存視角

那么如何分析 Flink 任務(wù)的堆內(nèi)存呢？這里我們是結(jié)合 Flink 任務(wù) GC 日志來進(jìn)行分析。GC 日志包含了每次 GC 堆內(nèi)不同區(qū)域內(nèi)存的變化和使用情況。同時(shí)根據(jù) GC 日志，也能夠獲取到一個(gè) Taskmanager 每次 Full GC 后，老年代剩余空間大小?？梢哉f，獲取實(shí)時(shí)任務(wù)的 GC 日志，使我們進(jìn)行實(shí)時(shí)任務(wù)內(nèi)存分析的前提。

GC 日志內(nèi)容分析，這里我們借助開源的 GC Viewer 工具來進(jìn)行具體分析，每次分析完，我們能夠獲取到 GC 相關(guān)指標(biāo)，下面是通過 GC Viewer 分析一次 GC 日志的部分結(jié)果：

上面通過 GC 日志分析出單個(gè) Flink Taskmanager 堆總大小、年輕代、老年代分配的內(nèi)存空間、Full GC 后老年代剩余大小等，當(dāng)然還有很多其他指標(biāo)，相關(guān)指標(biāo)定義可以去 Github 具體查看。

這里最重要的還是Full GC 后老年代剩余大小這個(gè)指標(biāo)，按照《Java 性能優(yōu)化權(quán)威指南》這本書 Java 堆大小計(jì)算法則，設(shè) Full GC 后老年代剩余大小空間為 M，那么對(duì)的大小建議 3 ~ 4倍 M，新生代為 1 ~ 1.5 倍 M，老年代應(yīng)為 2 ~ 3 倍 M，當(dāng)然，真實(shí)對(duì)內(nèi)存配置，你可以按照實(shí)際情況，將相應(yīng)比例再調(diào)大些，用以防止流量暴漲情形。

所以通過 Flink 任務(wù)的 GC 日志，我們可以計(jì)算出實(shí)時(shí)任務(wù)推薦的堆內(nèi)存總大小，當(dāng)發(fā)現(xiàn)推薦的堆內(nèi)存和實(shí)際實(shí)時(shí)任務(wù)的堆內(nèi)存大小相差過大時(shí)，我們就認(rèn)為能夠去降低業(yè)務(wù)方實(shí)時(shí)任務(wù)的內(nèi)存配置，從而降低機(jī)器內(nèi)存資源的使用。

1.2.2 任務(wù)消息處理能力視角

對(duì)于 Flink 任務(wù)消息處理能力分析，我們主要是看實(shí)時(shí)任務(wù)消費(fèi)的數(shù)據(jù)源單位時(shí)間的輸入，和實(shí)時(shí)任務(wù)各個(gè) Operator / Task 消息處理能力是否匹配。Operator 是 Flink 任務(wù)的一個(gè)算子，Task 則是一個(gè)或者多個(gè)算子 Chain 起來后，一起執(zhí)行的物理載體。

數(shù)據(jù)源我們內(nèi)部一般使用 Kafka，Kafka Topic 的單位時(shí)間輸入可以通過調(diào)用 Kafka Broker JMX 指標(biāo)接口進(jìn)行獲取，當(dāng)然你也可以調(diào)用 Flink Rest Monitoring 相關(guān) API 獲取實(shí)時(shí)任務(wù)所有 Kafka Source Task 單位時(shí)間輸入，然后相加即可。不過由于反壓可能會(huì)對(duì) Source 端的輸入有影響，這里我們是直接使用 Kafka Broker 指標(biāo) JMX 接口獲取 Kafka Topic 單位時(shí)間輸入。

在獲取到實(shí)時(shí)任務(wù) Kafka Topic 單位時(shí)間輸入后，下面就是判斷實(shí)時(shí)任務(wù)的消息處理能力是否與數(shù)據(jù)源輸入匹配。一個(gè)實(shí)時(shí)任務(wù)整體的消息處理能力，會(huì)受到處理最慢的 Operator / Task 的影響。打個(gè)比方，F(xiàn)link 任務(wù)消費(fèi)的 Kafka Topic 輸入為 20000 Record / S，但是有一個(gè) Map 算子，其并發(fā)度為 10 ，Map 算子中業(yè)務(wù)方調(diào)用了 Dubbo，一個(gè) Dubbo 接口從請(qǐng)求到返回為 10 ms，那么 Map 算子處理能力 1000 Record / S (1000 ms / 10 ms * 10 )，從而實(shí)時(shí)任務(wù)處理能力會(huì)下降為 1000 Record / S。

由于一條消息記錄的處理會(huì)在一個(gè) Task 內(nèi)部流轉(zhuǎn)，所以我們?cè)噲D找出一個(gè)實(shí)時(shí)任務(wù)中，處理最慢的 Task 邏輯。如果 Source 端到 Sink 端全部 Chain 起來的話，我們則是會(huì)找出處理最慢的 Operator 的邏輯。在源碼層，我們針對(duì) Flink Task 以及 Operator 增加了單條記錄處理時(shí)間的自定義 Metric，之后該 Metric 可以通過 Flink Rest API 獲取。我們會(huì)遍歷一個(gè) Flink 任務(wù)中所有的 Task , 查詢處理最慢的 Task 所在的 JobVertex（JobGraph 的點(diǎn)），然后獲取到該 JobVertex 所有 Task 的總輸出，最終會(huì)和 Kafka Topic 單位時(shí)間輸入進(jìn)行比對(duì)，判斷實(shí)時(shí)任務(wù)消息處理能力是否合理。

設(shè)實(shí)時(shí)任務(wù) Kafka Topic 單位時(shí)間的輸入為 S，處理最慢的 Task 代表的 JobVertex 的并發(fā)度為 P，處理最慢的 Task 所在的 JobVertex 單位時(shí)間輸出為 O，處理最慢的 Task 的最大消息處理時(shí)間為 T，那么通過下面邏輯進(jìn)行分析：

1. 當(dāng) O 約等于 S，且 1 second / T * P 遠(yuǎn)大于 S 時(shí)，會(huì)考慮減小任務(wù)并發(fā)度。

2. 當(dāng) O 約等于 S，且 1 second / T * P 約等于 S 時(shí)，不考慮調(diào)整任務(wù)并發(fā)度。

3. 當(dāng) O 遠(yuǎn)小于 S，且 1 second / T * P 遠(yuǎn)小于 S 時(shí)，會(huì)考慮增加任務(wù)并發(fā)度。

目前主要是 1 這種情況在 CPU 使用方面不合理，當(dāng)然，由于不同時(shí)間段，實(shí)時(shí)任務(wù)的流量不同，所以我們會(huì)有一個(gè)周期性檢測的的任務(wù)，如果檢測到某個(gè)實(shí)時(shí)任務(wù)連續(xù)多次都符合 1 這種情況時(shí)，會(huì)自動(dòng)報(bào)警提示平臺(tái)管理員進(jìn)行資源優(yōu)化調(diào)整。
下圖是從 Flink 任務(wù)的內(nèi)存以及消息處理能力兩個(gè)視角分析資源邏輯圖：

二、從內(nèi)存視角對(duì) Flink 分析實(shí)踐

2.1 Flink 任務(wù)垃圾回收器選擇

Flink 任務(wù)本質(zhì)還是一個(gè) Java 任務(wù)，所以也就會(huì)涉及到垃圾回收器的選擇。選擇垃圾回收器一般需要從兩個(gè)角度進(jìn)行參考：

1. 吞吐量，即單位時(shí)間內(nèi)，任務(wù)執(zhí)行時(shí)間 / (任務(wù)執(zhí)行時(shí)間 + 垃圾回收時(shí)間)，當(dāng)然并不是說降低 GC 停頓時(shí)間就能提升吞吐量，因?yàn)榻档?GC 停頓時(shí)間，你的 GC 次數(shù)也會(huì)上升。

2. 延遲。如果你的 Java 程序涉及到與外部交互，延遲會(huì)影響外部的請(qǐng)求使用體驗(yàn)。

Flink 任務(wù)我認(rèn)為還是偏重吞吐量的一類 Java 任務(wù)，所以會(huì)從吞吐量角度進(jìn)行更多的考量。當(dāng)然并不是說完全不考慮延遲，畢竟 JobManager、TaskManager、ResourceManager 之間存在心跳，延遲過大，可能會(huì)有心跳超時(shí)的可能性。

目前我們 JDK 版本為內(nèi)部 JDK 1.8 版本，新生代垃圾回收器使用 Parallel Scavenge，那么老年代垃圾回收器只能從 Serial Old 或者 Parallel Old 中選擇。由于我們 Flink k8s 任務(wù)每個(gè) Pod 的 CPU 限制為 0.6 - 1 core ，最大也只能使用 1 個(gè) core，所以老年代的垃圾回收器我們使用的是 Serial Old ，多線程垃圾回收在單 Core 之間，可能會(huì)有線程切換的消耗。

2.2 實(shí)時(shí)任務(wù) GC 日志獲取

設(shè)置完垃圾回收器后，下一步就是獲取 Flink 任務(wù)的 GC 日志。Flink 任務(wù)構(gòu)成一般是單個(gè) JobManager + 多個(gè) TaskManger ，這里需要獲取到 TaskManager 的 GC 日志進(jìn)行分析。那是不是要對(duì)所有 TaskManager 進(jìn)行獲取呢。這里我們按照 TaskManager 的 Young GC 次數(shù)，按照次數(shù)大小進(jìn)行排序，取排名前 16 的 TaskManager 進(jìn)行分析。YoungGC 次數(shù)可以通過 Flink Rest API 進(jìn)行獲取。

Flink on Yarn 實(shí)時(shí)任務(wù)的 GC 日志，直接點(diǎn)開 TaskManager 的日志鏈接就能夠看到，然后通過 HTTP 訪問，就能下載到本地。Flink On k8s 任務(wù)的 GC 日志，會(huì)先寫到 Pod 所掛載的云盤，基于 k8s hostpath volume 進(jìn)行掛載。我們內(nèi)部使用 Filebeat 進(jìn)行日志文件變更監(jiān)聽和采集，最終輸出到下游的 Kafka Topic。我們內(nèi)部會(huì)有自定義日志服務(wù)端，它會(huì)消費(fèi) Kafka 的日志記錄，自動(dòng)進(jìn)行落盤和管理，同時(shí)向外提供日志下載接口。通過日志下載的接口，便能夠下載到需要分析的 TaskManager 的 GC 日志。

2.3 基于 GC Viewer 分析 Flink 任務(wù)內(nèi)存

GC Viewer 是一個(gè)開源的 GC 日志分析工具。使用 GC Viewer 之前，需要先把 GC Viewer 項(xiàng)目代碼 clone 到本地，然后進(jìn)行編譯打包，就可以使用其功能。

在對(duì)一個(gè)實(shí)時(shí)任務(wù)堆內(nèi)存進(jìn)行分析時(shí)，先把 Flink TaskManager 的日志下載到本地，然后通過 GC Viewer 對(duì)日志進(jìn)行。如果你覺得多個(gè) Taskmanager GC 日志分析較慢時(shí)，可以使用多線程。上面所有這些操作，可以將其代碼化，自動(dòng)化產(chǎn)出分析結(jié)果。下面是通過 GC Viewer 分析的命令行：

java -jar gcviewer-1.37-SNAPSHOT.jar gc.log summary.csv

上面參數(shù) gc.log 表示一個(gè) Taskmanager 的 GC 日志文件名稱，summary.csv 表示日志分析的結(jié)果。下面是我們平臺(tái)對(duì)于某個(gè)實(shí)時(shí)任務(wù)內(nèi)存分析的結(jié)果：

下面是上面截圖中，部分參數(shù)說明：

1. RunHours，F(xiàn)link 任務(wù)運(yùn)行小時(shí)數(shù)

2. YGSize，一個(gè) TaskManager 新生代堆內(nèi)存最大分配量，單位兆

3. YGUsePC，一個(gè) TaskManager 新生代堆最大使用率

4. OGSize，一個(gè) TaskManager 老年代堆內(nèi)存最大分配量，單位兆

5. OGUsePC，一個(gè) TaskManager 老生代堆最大使用率

6. YGCoun，一個(gè) TaskMnager Young GC 次數(shù)

7. YGPerTime，一個(gè) TaskMnager Young GC 每次停頓時(shí)間，單位秒

8. FGCount，一個(gè) TaskMnager Full GC 次數(shù)

9. FGAllTime，一個(gè) TaskMnager Full GC 總時(shí)間，單位秒

10. Throught，Task Manager 吞吐量

11. AVG PT（分析結(jié)果 avgPromotion 參數(shù)），平均每次 Young GC 晉升到老年代的對(duì)象大小

12. Rec Heap，推薦的堆大小

13. RecNewHeap，推薦的新生代堆大小

14. RecOldHeap，推薦的老年代堆大小

上述大部分內(nèi)存分析結(jié)果，通過 GC Viewer 分析都能得到，不過推薦堆大小、推薦新生代堆大小、推薦老年代堆大小則是根據(jù) 1.2.1 小節(jié)的內(nèi)存優(yōu)化規(guī)則來設(shè)置。

三、從消息處理視角對(duì) Flink 分析實(shí)踐

3.1 實(shí)時(shí)任務(wù) Kafka Topic 單位時(shí)間輸入獲取

想要對(duì) Flink 任務(wù)的消息處理能力進(jìn)行分析，第一步便是獲取該實(shí)時(shí)任務(wù)的 Kafka 數(shù)據(jù)源 Topic，目前如果數(shù)據(jù)源不是 Kafka 的話，我們不會(huì)進(jìn)行分析。Flink 任務(wù)總體分為兩類：Flink Jar 任務(wù)和 Flink SQL 任務(wù)。Flink SQL 任務(wù)獲取 Kafka 數(shù)據(jù)源比較簡單，直接解析 Flink SQL 代碼，然后獲取到 With 后面的參數(shù)，再過濾掉 Sink 表之后，如果 SQLCreateTable 的 Conector 類型為 Kafka，就能夠通過 SQLCreateTable with 后的參數(shù)，拿到具體 Kafka Topic。

Flink Jar 任務(wù)的 Kafka Topic 數(shù)據(jù)源獲取相對(duì)繁瑣一些，我們內(nèi)部有一個(gè)實(shí)時(shí)任務(wù)血緣解析服務(wù)，通過對(duì) Flink Jar 任務(wù)自動(dòng)構(gòu)建其 PackagedProgram，PackagedProgram 是 Flink 內(nèi)部的一個(gè)類，然后通過 PackagedProgram ，我們可以獲取一個(gè) Flink Jar 任務(wù)的 StreamGraph，StreamGraph 里面有 Source 和 Sink 的所有 StreamNode，通過反射，我們可以獲取 StreamNode 里面具體的 Source Function，如果是 Kafka Source Sunction，我們就會(huì)獲取其 Kafka Topic。下面是 StreamGraph 類截圖：

獲取到 Flink 任務(wù)的 Kafka Topic 數(shù)據(jù)源之后，下一步便是獲取該 Topic 單位時(shí)間輸入的消息記錄數(shù)，這里可以通過 Kafka Broker JMX Metric 接口獲取，我們則是通過內(nèi)部 Kafka 管理平臺(tái)提供的外部接口進(jìn)行獲取。

3.2 自動(dòng)化檢測 Flink 消息處理最慢 Task

首先，我們?cè)谠创a層增加了 Flink Task 單條記錄處理時(shí)間的 Metric，這個(gè) Metric 可以通過 Flink Rest API 獲取。接下來就是借助 Flink Rest API，遍歷要分析的 Flink 任務(wù)的所有的 Task。Flink Rest Api 有這樣一個(gè)接口：

base_flink_web_ui_url/jobs/:jobid

這個(gè)接口能夠獲取一個(gè)任務(wù)的所有 Vertexs，一個(gè) Vertex 可以簡單理解為 Flink 任務(wù) JobGraph 里面的一個(gè) JobVertex。JobVertex 代表著實(shí)時(shí)任務(wù)中一段執(zhí)行邏輯。

獲取完 Flink 任務(wù)所有的 Vertex 之后，接下來就是獲取每個(gè) Vertex 具體 Task 處理單條記錄的 metric，可以使用下面的接口：

需要在上述 Rest API 鏈接 metrics 之后添加 ?get=（具體meitric ），比如：metrics?get=0.Filter.numRecordsOut，0 表示該 Vertex Task 的 id，F(xiàn)ilter.numRecordsOut 則表示具體的指標(biāo)名稱。我們內(nèi)部使用 taskOneRecordDealTime 表示Task 處理單條記錄時(shí)間 Metric，然后用 0.taskOneRecordDealTime 去獲取某個(gè) Task 的單條記錄處理時(shí)間的指標(biāo)。上面接口支持多個(gè)指標(biāo)查詢，即 get 后面使用逗號(hào)隔開即可。

最終自動(dòng)化檢測 Flink 消息處理最慢 Task 整體步驟如下：

1. 獲取一個(gè)實(shí)時(shí)任務(wù)所有的 Vertexs

2. 遍歷每個(gè) Vertex，然后獲取這個(gè) Vertex 所有并發(fā)度 Task 的 taskOneRecordDealTime，并且記錄其最大值

3. 所有 Vertex 單條記錄處理 Metric 最大值進(jìn)行對(duì)比，找出處理時(shí)間最慢的 Vertex。

下面是我們實(shí)時(shí)平臺(tái)對(duì)于一個(gè) Flink 實(shí)時(shí)任務(wù)分析的結(jié)果：

四、有贊 Flink 實(shí)時(shí)任務(wù)資源優(yōu)化實(shí)踐

既然 Flink 任務(wù)的內(nèi)存以及消息處理能力分析的方式已經(jīng)有了，那接下來就是在實(shí)時(shí)平臺(tái)端進(jìn)行具體實(shí)踐。我們實(shí)時(shí)平臺(tái)每天會(huì)定時(shí)掃描所有正在運(yùn)行的 Flink 任務(wù)，在任務(wù)內(nèi)存方面，我們能夠結(jié)合 實(shí)時(shí)任務(wù) GC 日志，同時(shí)根據(jù)內(nèi)存優(yōu)化規(guī)則，計(jì)算出 Flink 任務(wù)推薦的堆內(nèi)存大小，并與實(shí)際分配的 Flink 任務(wù)的堆內(nèi)存進(jìn)行比較，如果兩者相差的倍數(shù)過大時(shí)，我們認(rèn)為 Flink 任務(wù)的內(nèi)存配置存在浪費(fèi)的情況，接下來我們會(huì)報(bào)警提示到平臺(tái)管理員進(jìn)行優(yōu)化。

平臺(tái)管理員再收到報(bào)警提示后，同時(shí)也會(huì)判定實(shí)時(shí)任務(wù)消息能力是否合理，如果消息處理最慢的 Vertex (某段實(shí)時(shí)邏輯)，其所有 Task 單位時(shí)間處理消息記錄數(shù)的總和約等于實(shí)時(shí)任務(wù)消費(fèi)的 Kafka Topic 單位時(shí)間的輸入，但通過 Vertex 的并發(fā)度，以及單條消息處理 Metric ，算出該 Vertex 單位時(shí)間處理的消息記錄數(shù)遠(yuǎn)大于 Kafka Topic 的單位輸入時(shí)，則認(rèn)為 Flink 任務(wù)可以適當(dāng)調(diào)小并發(fā)度。具體調(diào)整多少，會(huì)和業(yè)務(wù)方溝通之后，在進(jìn)行調(diào)整。整體 Flink 任務(wù)資源優(yōu)化操作流程如下：

上述就是小編為大家分享的如何進(jìn)行Flink實(shí)時(shí)任務(wù)資源優(yōu)化探索與實(shí)踐了，如果剛好有類似的疑惑，不妨參照上述分析進(jìn)行理解。如果想知道更多相關(guān)知識(shí)，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。