一、背景
Flink目前是業界主流的流批一體的計算引擎。OPPO基于Flink建構的實時計算平台總共運作5000+作業，服務于公司二十幾條業務線。我們從2021年開始着手計算上雲的工作，目前已經有1/5的作業穩定運作在k8s上。
實時計算任務有以下特點：
- 資源初始固定。任務需要在送出之前明确資源用量且作業運作過程中不會自動調整
- 任務常駐。由于資料源多為無界流式資料。一旦有新的流資料進入任務，它會立刻發起并進行一次計算任務，是以整個過程是持續進行的；
- 負載呈周期性變化。實時作業流量和負載會随着時間的變化出現明顯的波峰波谷
由于以上特點帶來了以下問題：
- 調優困難。通常，使用者需要花費大量的時間進行作業調優。例如，新上線一個作業，需要考慮如何配置該作業的資源、并發數、TaskManager個數及大小等
- 資源用量無法比對負載的變化。由于實時作業的負載往往随着流量的變化而變化，初始設定的資源量容易過多或太少，進而造成資源浪費或者資源不足而導緻作業延時。而解決這個問題使用者往往需要手動重新開機作業再次設定資源用量，而這種操作繁瑣的同時也是滞後的
下圖展示的是某作業流量和資源使用情況：
由圖中看出作業呈明顯的波峰波谷的情況。
二、技術方案
2.1 整體架構
為了解決上述問題，我們設計實作了一種基于雲原生的Flink自動擴縮容的方案。整體方案以kubernete為基座，自研的"自動診斷 + 彈性伸縮"為核心。其中自動診斷子產品由 SmartResource 負責，彈性伸縮的能力承接由 Flink 計算架構負責。
- SmartResource負責根據作業上報的名額、使用者自定義的監控規則、全鍊路診斷規則來判斷目前作業的運作的健康度，作業是否需要伸縮均源于此。
- Flink 計算架構使用自研自适應排程器RescaleMonitor，自動感覺資源的變化，動态改變算子的并行度，并重新排程作業。
架構圖如下：
整體流程如下：
- 使用者配置自定義的伸縮政策到 SmartResource
- SmartResource 将相關的政策轉化為告警規則并配置到雲監控上
- 雲監控的資料來自 Flink 叢集上報的作業相關的各種名額
- 雲監控觸發告警時回調指定的 SmartResource 接口，SR 根據告警資訊，在任務鍊路打上優化建議Tag；此外，需要調整資源的，結合使用者配置的伸縮規則自動計算需要伸縮的副本數，并告知後端服務
- 後端服務收到伸縮請求後，再次将此請求在診斷聯調上應用一遍來确認是否是個有效的請求，如果是會在資料庫中存儲一個狀态為 pending 的伸縮請求
- 當作業的 Checkpoint 完成後，會通知 ostream backend 然後開始執行Flink計算資源的伸縮
2.2 方案詳述：
Flink 叢集往往由一個 JobManager 配合多個 TaskManager 構成。在縱向上也就是在單個 TaskManager的參數配置上，主要關注 CPU、記憶體兩個方面。在橫向上主要關注 TM 的數量。同時 TM 數量也代表了整個叢集可用 slot 個數，我們在伸縮的時候也是從縱橫向兩個方面考慮。
1. 縱向伸縮：
縱向的伸縮主要依賴于 Pod 在聲明資源的時候設定 request 和 limit。當我們在建立 TaskManager的時候，在使用者配置的基礎資源量上額外設定最小資源量(降低下限)，最大資源量會略大于使用者配置的基礎資源量(提高上限)。作業的負載波動的時候，單個 Pod(TM)占用的資源也會在 request 和 limit 之間波動。這樣在縱向上，減少資源的固定占用。也能很好的解決堆外記憶體占用突高引起的容器OOM的問題。
示例配置如下：
2. 橫向伸縮
橫向的伸縮主要表現在 TaskManager 數量的增減上，TaskManager的增減同樣反應整個叢集可用的資源數上(slot)，但是這裡兩個問題需要解決：
- 當 TaskManager增減的時候 JobManager 以及已經申請的 TaskManager不能丢失，也就是怎麼保持已申請的資源？
- 當 TaskManager增減的時候 JobManager 要快速感覺新增的 slot，也就是如何感覺資源變化并快速排程Job？
2.2.1 雲原生獨立部署模式
針對橫向伸縮第一個問題我們設計了基于 k8s 的獨立的作業部署式：kubernetes-standalone-application。此模式會獨立部署副本為1的JobManager，副本為n（根據使用者指定的并發自動計算）的TaskManager，兩者都是采用kubernetes deploy模式部署。由于TM的生命周期不在由JM控制，是以我們可以在外部控制TM的數量，這位我們後續的彈性伸縮打下了基礎。
此模式是我們使用雲原生的方式封裝了Flink自帶的standalone模式，在原生的standalone下，假如要部署JobManager、TaskManager的時候需要建構兩個非常複雜的的kubernetes yaml檔案(例如使用deploy)，在部署、平台化方面存在諸多不便之處。我們在原有的指令行基礎上支援了standalone的部署模式的支援，相容現有用戶端的各種參數指令。
新的部署模式圖如下：
特點如下：
- 雲原生一鍵部署；
- 自動推斷資源參數；
- 自研KSA Resourcemanager管理已申請資源；
- 與現有管理平台無縫內建
- 通過k8s的watch機制自動追蹤資源狀态；
部署指令如下：
2.2.2 資源伸縮協調器
針對橫向伸縮第二個問題我們設計了RescaleMonitor，它是一個伸縮螢幕，主要用于快速感覺資源的變化并決定是否需要重新排程作業。它首先計算作業所需的資源，計算完資源後，會一直等到可用資源穩定下來，一旦資源穩定，會開始确定作業的實際并行度, 一旦确定了并行度并且執行與可用槽比對，排程程式就會部署執行。
- 快速部署。隻要有外部資源增加的時候，RescaleMonitor會判斷是否滿足排程作業的最小資源，如果滿足就會立刻部署 jobgraph，不滿足會一直等待。
- 快速失敗。當有外部資源減少的時候，依賴于Kubernetes的watch機制可以快速感覺資源的變化，無需等待資源逾時。此處我們以pod名字作為Resource ID，可以在内部快速定位資源建立連接配接，進而執行各種操作。
- 安全恢複。當有外部資源增減的時候不會立即執行，Pending下來等待checkpoint完成的時候會檢查是否存在資源增減的Request，如果由才會立即執行重新部署JobGrap并從目前完成的checkpoint恢複，這樣在保證作業不丢的情況下，盡量減少重複消費資料的可能。
- 可固定并發。支援固定某些算子的最大并發，這在作業類似是消費 kafka 的時候特别有用。
- UI關聯。當資源變化的時候會實時通知web前端實時顯示最新的資源用量。
排程流程示意圖：
三、方案實踐及效果
3.1 彈性伸縮
我們在平台提供了使用者開啟彈性伸縮的前端頁面，也給出了常用的預設設定，如下圖所示，在CPU使用率大于70%并持續5分鐘的時候，開啟擴容，在CPU使用率小于30%并持續5分鐘的時候，開啟縮容。
彈性伸縮效果圖如下，通過途中可以看出，作業在cpu使用率低時，自動降低的并發。
我們記錄了每次伸縮的事件，包括時間、觸發原因，伸縮前後的資源等，友善平台跟使用者跟蹤資源的變化和排查問題。
3.2 自動診斷
以某作業為例，此次作業的存在明顯的波谷，且資源使用率很低。
通過我們的智能診斷服務可以自動優化整個作業的資源配比，在不重新開機作業的情況下完成TM的動态調整。
優化後的效果如下
經過我們計算經過自動診斷加彈性伸縮可以為業務帶來至少20%的成本降低。
其實成本降低隻是收益一方面，讓流量更加契合資源，讓診斷更加智能，讓作業穩定高效的運作，才能更好的服務于下遊業務。也是我們實時計算平台服務的宗旨。
四、後續規劃
後續我們在繼續優化彈性伸縮效果的基礎上，繼續朝着下面幾個方向努力。
- batch任務的自适應排程
- 基于機器學習的自動化診斷
- 流批一體的錯峰排程
附錄
- https://nightlies.apache.org/flink/flink-docs-master/zh/docs/deployment/elastic_scaling/
- https://cwiki.apache.org/confluence/display/FLINK/FLIP-159%3A+Reactive+Mode
- https://cwiki.apache.org/confluence/display/FLINK/FLIP-160%3A+Adaptive+Scheduler
- https://kubernetes.io/zh-cn/docs/tasks/run-application/horizontal-pod-autoscale/
作者簡介
John，OPPO實時計算平台進階研發工程師，Apache Flink Contributor，長期專注于大資料計算領域。
來源:微信公衆号:OPPO數智技術
出處:https://mp.weixin.qq.com/s?__biz=Mzg4MzE2MzY1OA==&mid=2247489406&idx=1&sn=c81313bd1e7c70f6160aa02addef8f60