摘要：本文整理自小米大數(shù)據(jù)部高級(jí)軟件工程師張蛟在 Flink Forward Asia 2021 生產(chǎn)實(shí)踐專(zhuān)場(chǎng)的演講。主要內(nèi)容包括：

發(fā)展現(xiàn)狀和規(guī)模
穩(wěn)定性?xún)?yōu)化及實(shí)踐
運(yùn)維優(yōu)化及實(shí)踐
未來(lái)規(guī)劃與展望

Tips：點(diǎn)擊「閱讀原文」查看原文視頻&PPT～

發(fā)展現(xiàn)狀及規(guī)模

現(xiàn)階段，我們的整體架構(gòu)可以分成5層，數(shù)據(jù)從下往上流動(dòng)，如上圖。

數(shù)據(jù)采集層主要負(fù)責(zé)收集各類(lèi)數(shù)據(jù)，數(shù)據(jù)的來(lái)源分為兩類(lèi)，一類(lèi)是埋點(diǎn)和業(yè)務(wù)日志以及服務(wù)日志，經(jīng)由 LCS Agent 進(jìn)行采集，另一類(lèi)是數(shù)據(jù)庫(kù)數(shù)據(jù)經(jīng)由 Binlog 或 Checkpoint 數(shù)據(jù)集成等方式收集到消息隊(duì)列中。以 Flink、Spark 為主的計(jì)算層對(duì)其進(jìn)行處理，并最終存儲(chǔ)到各類(lèi)存儲(chǔ)和查詢(xún)服務(wù)中，供業(yè)務(wù)使用。Flink 是計(jì)算層實(shí)時(shí)和準(zhǔn)實(shí)時(shí)處理的主要框架，在其中正發(fā)揮著越來(lái)越重要的作用，尤其是 Flink+Iceberg 數(shù)據(jù)湖技術(shù)，正在讓流批一體成為現(xiàn)實(shí)。

目前我們的集群上運(yùn)行著 3000 多個(gè)作業(yè)，主力版本是 1.12，1.14 版本也已經(jīng)合并上線(xiàn)，日均處理 10 萬(wàn)億+ 條消息，PB 級(jí)的數(shù)據(jù)量，峰值數(shù)據(jù) 2 億條/秒，運(yùn)行在國(guó)內(nèi)外 10 多個(gè)集群，使用超過(guò) 45000 個(gè) CPU core，內(nèi)存使用超過(guò) 200tb。

在這樣規(guī)模的數(shù)據(jù)處理過(guò)程中，我們遇到了許多問(wèn)題。

作業(yè)內(nèi)存占用不可控，on Yarn 模式非常容易出現(xiàn) Yarn container OOM kill，導(dǎo)致 container lost，引發(fā)作業(yè)頻繁重啟，包括框架內(nèi)重啟。
on Yarn 模式無(wú)法支持作業(yè)自動(dòng)平滑重啟，在機(jī)器過(guò)保、下線(xiàn)、機(jī)房遷移等過(guò)程中，只能觸發(fā) failover。
實(shí)時(shí)作業(yè)對(duì)負(fù)載較為敏感，啟動(dòng)和運(yùn)行的過(guò)程中需要保證機(jī)器性能，避免因離線(xiàn)和在線(xiàn)混部造成影響。
Checkpoint 作為 Flink 有狀態(tài)計(jì)算數(shù)據(jù)一致性的保障，存在穩(wěn)定性問(wèn)題。
historyserver 默認(rèn)的清理策略不好設(shè)置，導(dǎo)致占用的磁盤(pán)空間比較大，訪問(wèn)慢。
作業(yè)異常時(shí)難以確定異常原因和節(jié)點(diǎn)，需要查看大量的作業(yè)日志，導(dǎo)致故障排查困難。

穩(wěn)定性?xún)?yōu)化及實(shí)踐

首先是 Yarn container lost 的優(yōu)化。Flink JobManager 首先會(huì)向 Yarn reCheckpointmanager 申請(qǐng)資源，Yarn reCheckpointmanager 為該申請(qǐng)分配資源后將分配信息返回給 JobManager，然后 JobManager 會(huì)根據(jù)分配信息去啟動(dòng) taskmanager，并使之與 JobManager 進(jìn)行心跳。

JobManager 包括 JobMaster 和 reCheckpointmanager，它會(huì)主動(dòng)發(fā)送心跳請(qǐng)求，探測(cè) taskmanager 是否存活。如果 taskexecutor 因?yàn)槟承┰蛞馔獗?kill，JobManager 的日志中就會(huì)提示 container lost。

上圖是 container lost 現(xiàn)象的提示之一，一般老版本的 Flink 中出現(xiàn)比較多。

上圖是 container lost 現(xiàn)象的另一種提示。

在出現(xiàn) container lost 時(shí)，如果去查看 Yarn的nodemanager 或 JobManager 中異常前后的日志，一般都可以看到類(lèi)似 beyond the physical memory limit 的日志，這表明它是因?yàn)槲锢韮?nèi)存使用超限被 Yarn kill。

這里需要先介紹一下 Yarn 控制內(nèi)存超用的方式，Yarn Nodemanager 會(huì)啟動(dòng)一個(gè) containersmonitor 的線(xiàn)程，這個(gè)線(xiàn)程會(huì)定期掃描 Nodemanager 上的 container 內(nèi)存占用，從而實(shí)現(xiàn)內(nèi)存資源的隔離。

簡(jiǎn)單來(lái)說(shuō)，如果某個(gè) container 對(duì)應(yīng)進(jìn)程樹(shù)中所有年齡大于 0 的進(jìn)程，總內(nèi)存使用量超過(guò)申請(qǐng)量的兩倍，或所有年齡大于 1 的進(jìn)程，總內(nèi)存使用量超過(guò)上限，就表明其內(nèi)存超用，需要被 kill。

但實(shí)際上這種方式存在一定的問(wèn)題：

一是定期掃描對(duì)于內(nèi)存突增的隔離性比較差，可能還沒(méi)有開(kāi)始掃描就已經(jīng)達(dá)到系統(tǒng)總內(nèi)存上限，導(dǎo)致被系統(tǒng) kill；
二是 Yarn 通常會(huì)開(kāi)啟節(jié)點(diǎn)資源的超賣(mài)，此時(shí)如果所有資源都被使用，會(huì)導(dǎo)致節(jié)點(diǎn)不穩(wěn)定；
三是如果作業(yè)只是臨時(shí)的內(nèi)存需求，即使此時(shí)節(jié)點(diǎn)仍有富余內(nèi)存，也會(huì)觸發(fā) kill。

針對(duì)這些問(wèn)題，我們采用 Cgroup + JDK升級(jí) + Jemalloc 的方式進(jìn)行了優(yōu)化?？赡苡腥藭?huì)問(wèn)為什么需要進(jìn)行 JDK 升級(jí)？這是因?yàn)槔习姹镜?JDK 使用 Jemalloc 存在線(xiàn)程死鎖的問(wèn)題，另外升級(jí)最新的 JDK 也能避免其他的 JDK bug，通常這類(lèi) bug 都不容易被找到和復(fù)現(xiàn)。

Cgroup 的方式主要是開(kāi)啟內(nèi)存軟限制，它對(duì) container 的內(nèi)存限制不再是基于單個(gè) container 的內(nèi)存申請(qǐng)量，而是整個(gè) Nodemanager 的內(nèi)存量。這個(gè)時(shí)候如果 NodeManager 上仍有富余內(nèi)存，內(nèi)存超用的 container 就可以接著使用這些富余的內(nèi)存。一個(gè)節(jié)點(diǎn)上同時(shí)存在多個(gè) Container 內(nèi)存超用導(dǎo)致整個(gè)節(jié)點(diǎn)內(nèi)存達(dá)到上限，才會(huì)觸發(fā) oom event。Oom listener 對(duì)該事件進(jìn)行監(jiān)聽(tīng)并判斷，如果達(dá)到節(jié)點(diǎn)總內(nèi)存就會(huì)選取內(nèi)存實(shí)際占用量超過(guò)申請(qǐng)量且啟動(dòng)時(shí)間最短、優(yōu)先級(jí)最低的作業(yè)觸發(fā) oom kill。

然而，Cgroup 只是在一定程度上解決了 container 頻繁被 Yarn oom kill 導(dǎo)致 lost 的問(wèn)題，并沒(méi)有完全徹底地解決。在使用的過(guò)程中，依然存在某些 container 的內(nèi)存使用持續(xù)上漲，最終被 cgroup oom kill 的情況，然后我們發(fā)現(xiàn)該問(wèn)題可能與 glibc 的內(nèi)存分配 bug 有關(guān)，長(zhǎng)期運(yùn)行的進(jìn)程會(huì)存在連續(xù)多塊大小為 65536 的 anon 塊，所以我們最終的解決方案如下：

使用 Cgroup 解決內(nèi)存臨時(shí)超用的問(wèn)題，比如 RocksDB 對(duì)內(nèi)存的限制不嚴(yán)格、小白用戶(hù)對(duì)內(nèi)存的設(shè)置和使用不正確等造成的問(wèn)題，然后升級(jí) JDK 版本，解決 Jemalloc 分配時(shí)的線(xiàn)程死鎖 bug，最后切換 Jemalloc，解決 Linux 系統(tǒng)下的 64M anon 分配 bug。

經(jīng)過(guò)一系列的優(yōu)化，從上圖可以看出，container lost 的頻率由每月的近 5000 次減少到不到 100 次，因 Yarn oom kill 造成的作業(yè)異常重啟減少了 90% 以上，效果顯著。

第二個(gè)優(yōu)化實(shí)踐是節(jié)點(diǎn)的平滑重啟功能，流式作業(yè)是長(zhǎng)時(shí)間運(yùn)行的作業(yè)，由于大部分都運(yùn)行在廉價(jià)的機(jī)器上，因此機(jī)器出現(xiàn)過(guò)保、硬件故障、維修下線(xiàn)、機(jī)房遷移等都比較常見(jiàn)。為了提前預(yù)防可能出現(xiàn)的隱患，避免框架重啟造成的影響，提升云環(huán)境下作業(yè)的穩(wěn)定性，解決 Yarn 模式下恢復(fù)時(shí)間過(guò)長(zhǎng)帶來(lái)的問(wèn)題，我們開(kāi)發(fā)了作業(yè)的平滑重啟功能。

將節(jié)點(diǎn)加入到 exclude 后，F(xiàn)link recheckpoint manager 會(huì)獲取到 decommission 的信息，通過(guò)解析該信息得到對(duì)應(yīng)的節(jié)點(diǎn)，并判斷當(dāng)前運(yùn)行任務(wù)的 container 是否運(yùn)行在被 decommission 的節(jié)點(diǎn)上。如果是，就通過(guò)調(diào)用任務(wù)的 JobManager 的 stop with savepoint 接口去停止。平臺(tái)會(huì)自動(dòng)檢測(cè)任務(wù)的運(yùn)行狀態(tài)，如果某個(gè)作業(yè)不是通過(guò)平臺(tái)停止，則平臺(tái)會(huì)自動(dòng)將該任務(wù)重新拉起，作業(yè)從 savepoint 恢復(fù)。這個(gè)過(guò)程會(huì)進(jìn)行周期性的觸發(fā)并批量合并后再處理，避免消息頻繁觸發(fā)造成瞬時(shí)負(fù)載壓力。此外，節(jié)點(diǎn)和 container 都會(huì)進(jìn)行去重，避免對(duì)同一任務(wù)多次觸發(fā)影響穩(wěn)定性。另外它的觸發(fā)周期遠(yuǎn)小于 sre 在下線(xiàn)節(jié)點(diǎn)時(shí)設(shè)置的下線(xiàn)周期，也緩解了運(yùn)維壓力。

JobManager 會(huì)啟動(dòng)指標(biāo)收集監(jiān)控線(xiàn)程，并周期性地采集節(jié)點(diǎn)的 CPU、內(nèi)存、磁盤(pán) io 和網(wǎng)絡(luò) io 等指標(biāo)，然后匯聚成指標(biāo)集合，通過(guò)動(dòng)態(tài)指標(biāo)規(guī)則對(duì)指標(biāo)進(jìn)行判定，如果滿(mǎn)足條件就會(huì)將其加入到節(jié)點(diǎn)黑名單，這樣該 Application 的 container 便不會(huì)再運(yùn)行在這個(gè)節(jié)點(diǎn)上。如果某個(gè)節(jié)點(diǎn)被多個(gè) application 加入黑名單，則表明該節(jié)點(diǎn)可能存在問(wèn)題，會(huì)自動(dòng)觸發(fā)作業(yè)平滑重啟，并進(jìn)行監(jiān)控報(bào)警，以此來(lái)自動(dòng)發(fā)現(xiàn)可能的異常節(jié)點(diǎn)。

上圖是 Flink Checkpoint 的大致流程，Checkpoint coordinator 會(huì)觸發(fā) Checkpoint Operator 進(jìn)行 Checkpoint，Checkpoint Operator 生成并向下游廣播 Checkpoint Barrier，然后 Snapshot State。Checkpoint Operator 完成 Checkpoint 后進(jìn)行 ack，下游節(jié)點(diǎn)收到 Checkpoint Barrier 后，根據(jù)是否要進(jìn)行對(duì)齊做對(duì)應(yīng)的處理，然后進(jìn)入 Checkpoint 邏輯。所有的節(jié)點(diǎn)都向 Checkpoint Coordinateor ack 之后，表示該次 Checkpoint 已經(jīng)完成，接著向所有參與 Checkpoint 的 Operator 發(fā)送完成通知，最后 Operator 做最后的提交操作等。

Checkpoint 過(guò)程中遇到的問(wèn)題包括以下這些：

磁盤(pán)滿(mǎn)或其他 io 異常，會(huì)導(dǎo)致 Checkpoint 長(zhǎng)期無(wú)法觸發(fā)，但異常信息只存在于 JobManager 的日志中，并不影響作業(yè)的正常執(zhí)行，導(dǎo)致潛在的隱患不容易被感知。
作業(yè)因邏輯變更、調(diào)整并發(fā)、重新調(diào)度等原因，重啟時(shí)默認(rèn)不會(huì)從 Checkpoint 恢復(fù)，導(dǎo)致?tīng)顟B(tài)丟失或者消息積壓。
大并發(fā)度時(shí) Checkpoint 小文件過(guò)多，引發(fā)大量的 HDFS RPC 負(fù)載壓力。
用戶(hù)錯(cuò)誤配置 Checkpoint 目錄引發(fā)的恢復(fù)沖突非常不容易控制，也不易排查。

針對(duì)以上問(wèn)題，我們也進(jìn)行了一些優(yōu)化。

針對(duì)磁盤(pán)滿(mǎn)、io 異常、Kerberos 文件損壞的問(wèn)題，我們會(huì)捕獲異常棧，根據(jù)異常棧進(jìn)行判斷和重試，并在失敗時(shí)增加 Checkpoint 的失敗計(jì)數(shù)，超過(guò)一定次數(shù)則進(jìn)行框架內(nèi)的重啟，或向用戶(hù)發(fā)送告警，保證作業(yè)不會(huì)出現(xiàn)長(zhǎng)時(shí)間的 Checkpoint 失敗而從一個(gè)非常老的 Checkpoint 恢復(fù)。

針對(duì)作業(yè)重啟時(shí)無(wú)法從 Checkpoint 恢復(fù)的問(wèn)題，優(yōu)化方式是對(duì)每個(gè)作業(yè)設(shè)置默認(rèn)的保留數(shù)量，并在進(jìn)行 Checkpoint 時(shí)先生成一個(gè)臨時(shí)的 Checkpoint Metadata 文件，只有在 Finalize 時(shí)才會(huì)被 rename 成正式的文件。接著將所有 Checkpoint 文件按最后修改時(shí)間降序排序，在其中尋找正式的 Checkpoint Metadata 文件。如果成功則表明其是一個(gè)完備的、可用于恢復(fù)的 Checkpoint 文件。

在這樣的設(shè)定下，必須確保文件最后修改時(shí)間的正確性。為此我們?cè)O(shè)置了任務(wù) finish 默認(rèn)不刪除 Checkpoint 文件，任務(wù)在做 Savepoint 時(shí)默認(rèn)不 discard 最新的 Checkpoint 文件，以確保這兩類(lèi)文件最后修改時(shí)間的正確性。通過(guò)以上方式保證了任務(wù)能自動(dòng)從最新的、完備的狀態(tài)進(jìn)行恢復(fù)，需要重新處理的數(shù)據(jù)和狀態(tài)盡量少。另外，如果任務(wù)已經(jīng)找到最新的、完備的 Checkpoint 并可以用來(lái)恢復(fù)，這表明前面的 Savepoint 和 Checkpoint 已經(jīng)可以清理，由此減少空間的占用。

于是我們通過(guò)為 Savepoint 設(shè)置生命周期來(lái)清理全量 Savepoint；對(duì)于增量的 Checkpoint，為了避免清除掉正在使用的狀態(tài)，會(huì)先去讀取其 Metadata 文件的內(nèi)容，將其中用到的狀態(tài)文件對(duì)應(yīng)的父文件夾保留，其余的進(jìn)行清理，從而確保在不影響狀態(tài)恢復(fù)的前提下，盡量減少文件數(shù)和空間占用。

針對(duì)用戶(hù)隨意配置 Checkpoint 目錄導(dǎo)致?tīng)顟B(tài)恢復(fù)沖突和引發(fā)負(fù)載壓力的問(wèn)題，通過(guò)在 Metadata 文件中增加作業(yè)名和時(shí)間戳，當(dāng)前的作業(yè)名與存儲(chǔ)的作業(yè)名不同則會(huì)提示告警信息，恢復(fù)的 Checkpoint 的時(shí)間戳與當(dāng)前時(shí)間存在較大的差異，也會(huì)有告警信息。

小文件是使用 HDFS 經(jīng)常遇到的問(wèn)題，由于 HDFS 適合于存儲(chǔ)大塊文件，所以必須對(duì)小文件進(jìn)行優(yōu)化來(lái)提升性能和穩(wěn)定性。方法是在進(jìn)行 Checkpoint 時(shí)對(duì)小文件寫(xiě)入進(jìn)行合并，比如將多個(gè)小文件寫(xiě)入到 sequence file 中，形成一個(gè)大的文件，這可能會(huì)造成空間浪費(fèi)，但是對(duì)降低 HDFS Namenode 負(fù)載壓力效果比較明顯。

此外通過(guò)聯(lián)邦集群的方式，使用多個(gè) Namenode 均衡 RPC 請(qǐng)求負(fù)載，每一個(gè) Namenode 都是一個(gè)相對(duì)獨(dú)立的服務(wù)，然后對(duì)用戶(hù)作業(yè)規(guī)范其 Checkpoint 目錄，使其訪問(wèn)能夠被均衡到多個(gè) Namenode 上，再對(duì)舊的 HDFS 文件通過(guò)掛載表的形式讀舊寫(xiě)新，逐步實(shí)現(xiàn)自動(dòng)遷移到新的統(tǒng)一的規(guī)范目錄下。

接下來(lái)介紹一個(gè)案例，該案例來(lái)自小米數(shù)據(jù)采集服務(wù)，圖示是他們非常簡(jiǎn)單的架構(gòu)圖，主要是將多個(gè)源端 SDK 的埋點(diǎn)和數(shù)據(jù)收集到消息隊(duì)列中，然后使用 Flink 進(jìn)行 ETL，最終存儲(chǔ)到 Doris 中并在看板上進(jìn)行展示。

目前該業(yè)務(wù)已經(jīng)接入 750+ 國(guó)內(nèi)外業(yè)務(wù)，日均處理 1600億+ 條消息。通過(guò)采用 Checkpoint 相關(guān)的優(yōu)化手段，將 RPC 延遲降低了約 40%，減少了小文件。同時(shí)在作業(yè)通過(guò) stop with savepoint 啟停時(shí)，保證了恢復(fù)的正確性，確保了 exactly once 的語(yǔ)義。

運(yùn)維優(yōu)化實(shí)踐

Flink Historyserver 對(duì)作業(yè)運(yùn)維非常有效，尤其是它能在作業(yè)停止后，查看作業(yè)的統(tǒng)計(jì)信息，如果作業(yè)異常退出或處理結(jié)果有問(wèn)題，我們又因?yàn)橐恍┰驘o(wú)法及時(shí)查看相關(guān)日志，就可以在將來(lái)通過(guò) Historyserver 查看。

Flink Historyserver 會(huì)在每一次定時(shí)清理時(shí)獲取上一次清理已經(jīng)被緩存的作業(yè) ID，再獲取本次已經(jīng)打包的歷史日志信息，然后判斷歷史日志是否已經(jīng)超過(guò)了配置的最大值，若是，就會(huì)將后面的歷史日志直接執(zhí)行清理，否則就會(huì)判斷上一次緩存的作業(yè)在當(dāng)次歷史日志中是否存在，如果不存在也會(huì)執(zhí)行清理。

但上述流程存在一系列的問(wèn)題，一個(gè)是服務(wù)重啟會(huì)造成當(dāng)前緩存的已下載的作業(yè)信息丟失，如果在重啟之間該作業(yè)的歷史日志也丟失，就會(huì)形成懸浮的緩存作業(yè)，本地緩存的作業(yè)將會(huì)長(zhǎng)期存在，無(wú)法清理。當(dāng)前已打包的歷史日志信息不支持過(guò)期，導(dǎo)致大量的日志存留于 HDFS 和本地磁盤(pán)，且會(huì)長(zhǎng)期存在，不僅影響訪問(wèn)的速度，也會(huì)造成磁盤(pán)空間的較大浪費(fèi)。緩存下來(lái)的作業(yè)歷史日志最大值難以確定，基礎(chǔ)服務(wù)如 HDFS 等如果出現(xiàn)異常，會(huì)導(dǎo)致同時(shí)出現(xiàn)大量失敗，沖走有效日志。另外當(dāng)前默認(rèn)并沒(méi)有記錄 Taskmanager 上的日志，非常不利于異常排查，

針對(duì)上述問(wèn)題我們也做了相應(yīng)優(yōu)化。

一個(gè)是讀取當(dāng)前已經(jīng)緩存到本地磁盤(pán)的作業(yè)歷史日志信息，并將其與歷史日志記錄進(jìn)行對(duì)比，從而避免出現(xiàn)懸浮的緩存作業(yè)；支持歷史日志的最長(zhǎng)保留時(shí)間，超過(guò)其生命周期就會(huì)進(jìn)行清理，相比于當(dāng)前支持的歷史日志最大保留數(shù)量，更加科學(xué)合理；另外我們也支持了 Taskmanager 和 Container 歷史數(shù)據(jù)的打包和清理，更全面地記錄作業(yè)在異常退出時(shí)的各項(xiàng)信息，方便排查問(wèn)題。

作業(yè)的全鏈路心跳監(jiān)控功能主要是對(duì)作業(yè)的鏈路延時(shí)進(jìn)行監(jiān)控，實(shí)現(xiàn)方式是通過(guò)在 Stream Checkpoint 中插入特殊標(biāo)記，標(biāo)記信息包括作業(yè)的名稱(chēng)、當(dāng)前的時(shí)間，名稱(chēng)的生成方式是 op+operator 在整個(gè)鏈路的 index 以及 subtask 在 operator 的 index，非 Checkpoint 節(jié)點(diǎn)會(huì)在收到標(biāo)記后更新名稱(chēng)，并用當(dāng)前的時(shí)間減去 Checkpoint 插入的時(shí)間生成從 Checkpoint 到該 subtask 的耗時(shí)，并上報(bào)到 Metrics Reporter 中，最終對(duì)這些 metrics 進(jìn)行計(jì)算，通過(guò)這種方式可以發(fā)現(xiàn)鏈路中的異常節(jié)點(diǎn)，監(jiān)測(cè)作業(yè)的數(shù)據(jù)異常丟失，還能夠通過(guò)心跳信息的插入頻率預(yù)估其影響。

心跳標(biāo)記在遇到多個(gè)下游鏈路時(shí)并不是隨機(jī)選擇鏈路，而是同時(shí)廣播到多條鏈路中，因此可能會(huì)出現(xiàn)心跳監(jiān)控標(biāo)記信息過(guò)多的情況，影響正常作業(yè)的處理性能。

這時(shí)就出現(xiàn)了一個(gè)矛盾點(diǎn)，全鏈路心跳監(jiān)控采樣越頻繁，對(duì)各節(jié)點(diǎn)處理性能的監(jiān)控就越及時(shí)準(zhǔn)確，但同時(shí)也會(huì)造成信息過(guò)多、影響正常數(shù)據(jù)的處理。針對(duì)這個(gè)問(wèn)題，我們進(jìn)行了以下三個(gè)方面的改進(jìn)和處理：

一是將 chain operator metrics 信息進(jìn)行合并上報(bào)，因?yàn)樗谋O(jiān)控信息基本相同，這樣可以減少上報(bào)的數(shù)據(jù)量。
二是通過(guò) restful 接口動(dòng)態(tài)啟停監(jiān)控，這樣只有在有異常時(shí)才會(huì)進(jìn)行采樣和監(jiān)控，正常情況下不影響作業(yè)的運(yùn)行。
三是通過(guò)對(duì)采樣進(jìn)行周期性的合并和處理，實(shí)現(xiàn)了對(duì)任務(wù) pipeline 數(shù)據(jù)量和延遲的預(yù)估以及監(jiān)控功能。

restful 接口動(dòng)態(tài)啟停監(jiān)控功能不僅能動(dòng)態(tài)啟停心跳監(jiān)控，我們發(fā)現(xiàn)還有其他場(chǎng)景也能從這個(gè)功能中受益，因此我們對(duì)其進(jìn)行了擴(kuò)展。簡(jiǎn)單的代碼修改就能讓它支持其他配置的動(dòng)態(tài)調(diào)整，包括 Checkpoint 配置，如 Checkpoint 周期和超時(shí)時(shí)間，動(dòng)態(tài)日志的級(jí)別等。

當(dāng)作業(yè)出現(xiàn)性能或 Checkpoint 問(wèn)題時(shí)，可以通過(guò) restful 接口動(dòng)態(tài)開(kāi)啟、問(wèn)題確定后動(dòng)態(tài)停止，這樣就能解決心跳信息過(guò)多的問(wèn)題。在負(fù)載突增、短時(shí)數(shù)據(jù)傾斜導(dǎo)致 Checkpoint 超時(shí)，動(dòng)態(tài)調(diào)整 Checkpoint 超時(shí)時(shí)間能避免作業(yè)因 Checkpoint 超時(shí)而失敗，它也能避免由于 Checkpoint 長(zhǎng)時(shí)間不成功導(dǎo)致數(shù)據(jù)積壓更多、數(shù)據(jù)傾斜問(wèn)題更嚴(yán)重而陷入的死循環(huán)。同時(shí)它還能用于確定超時(shí)時(shí)間，用戶(hù)可以通過(guò)動(dòng)態(tài)調(diào)整的方式，不斷測(cè)試最適合作業(yè)的超時(shí)時(shí)間，減少了壓測(cè)過(guò)程中的作業(yè)啟停次數(shù)。它也支持其他配置的調(diào)整，比如動(dòng)態(tài)調(diào)整日志級(jí)別，但是需要注意的是調(diào)整后的配置并沒(méi)有持久化，會(huì)因?yàn)榭蚣苤貑⒒蜃鳂I(yè)的重啟而失效。

未來(lái)規(guī)劃

未來(lái)，我們將在以下方面繼續(xù)探索：

持續(xù)開(kāi)發(fā)并優(yōu)化自動(dòng)彈性伸縮容的功能。Flink1.13 開(kāi)始提供了自動(dòng)彈性伸縮容的功能，但是目前并不完善，要在生產(chǎn)環(huán)境上用起來(lái)還需要做不少的工作。
版本收斂是很多Flink開(kāi)發(fā)人員都會(huì)遇到的一個(gè)問(wèn)題。Flink 社區(qū)的發(fā)展比較快，版本的發(fā)布和迭代也是非?？?。為了降低運(yùn)維壓力，緊跟社區(qū)，這也是勢(shì)在必行的。
對(duì) state 讀寫(xiě)性能進(jìn)行優(yōu)化，提升大狀態(tài)作業(yè)的性能。
Heartbeat timeout 也是目前線(xiàn)上對(duì)穩(wěn)定性影響比較大的問(wèn)題，我們也會(huì)進(jìn)行跟進(jìn)和優(yōu)化。
對(duì)作業(yè)啟動(dòng)和恢復(fù)性能進(jìn)行優(yōu)化，減少作業(yè)因各種原因造成的斷流是 Flink 社區(qū)和許多業(yè)務(wù)非常關(guān)注的問(wèn)題，我們同樣也有面臨著這樣的壓力。
繼續(xù)打磨批流融合能力，完善對(duì) batch 模式和數(shù)據(jù)湖等的支持，也是現(xiàn)在的熱點(diǎn)，我們希望能在這上面進(jìn)行更多探索，從而更好地支撐業(yè)務(wù)，也讓 Flink 的應(yīng)用更加廣泛。