編輯整理：趙明明
出品平臺：DataFunTalk
導(dǎo)讀：Flink是目前流式處理領(lǐng)域的熱門引擎，具備高吞吐、低延遲的特點，在實時數(shù)倉、實時風(fēng)控、實時推薦等多個場景有著廣泛的應(yīng)用。京東于2018年開始基于Flink+K8s深入打造高性能、穩(wěn)定、可靠、易用的實時計算平臺，支撐了京東內(nèi)部多條業(yè)務(wù)線平穩(wěn)度過618、雙11多次大促。本次講演將分享京東Flink計算平臺在容器化實踐過程中遇到的問題和方案，在性能、穩(wěn)定性、易用性等方面對社區(qū)版Flink所做的深入的定制和優(yōu)化，以及未來的展望和規(guī)劃。主要內(nèi)容包括：
實時計算演進(jìn)
Flink容器化實踐
Flink優(yōu)化改進(jìn)
未來規(guī)劃
最初大數(shù)據(jù)的模式基本都是T+1，但是隨著業(yè)務(wù)發(fā)展，對數(shù)據(jù)實時性的要求越來越高，比如對于一個數(shù)據(jù)，希望能夠在分鐘級甚至秒級得到計算結(jié)果。京東是在2014年開始基于Storm打造第一代流式計算平臺，并在Storm的基礎(chǔ)上，做了很多優(yōu)化改進(jìn)，比如基于cgroup實現(xiàn)對worker使用資源的隔離、網(wǎng)絡(luò)傳輸壓縮優(yōu)化、引入任務(wù)粒度toplogy master分擔(dān)zk壓力等。到2016年，Storm已經(jīng)成為京東內(nèi)部流式處理的最主要的計算引擎，服務(wù)于各個業(yè)務(wù)線，可以達(dá)到比較高的實時性。
隨著業(yè)務(wù)規(guī)模的不斷擴大，Storm也暴露出許多問題，特別是對于吞吐量巨大、但是對于延遲不是那么敏感的業(yè)務(wù)場景顯得力不從心。于是，京東在2017年引入了Spark Streaming流式計算引擎，用于滿足此類場景業(yè)務(wù)需要。
隨著業(yè)務(wù)的發(fā)展，不光是對于數(shù)據(jù)的延遲有很高要求，同時對于數(shù)據(jù)的吞吐處理能力也有很高的要求，所以迫切需要一個兼具低延遲和高吞吐能力的計算框架，于是在2018年我們引入了Flink。在Flink社區(qū)版的基礎(chǔ)上，我們從性能、穩(wěn)定性、易用性還有功能等方面，都做了一些深入的定制和優(yōu)化。同時我們基于k8s實現(xiàn)了實時計算全面的容器化，因為容器化有很多的優(yōu)點，它可以做到很好的資源隔離，同時它有一個很強的自愈能力，另外它很容易實現(xiàn)資源的彈性調(diào)度。同時我們基于Flink打造了全新的SQL平臺，降低用戶開發(fā)實時計算應(yīng)用的門檻。
到2020年，基于Flink和k8s實時計算平臺已經(jīng)做的比較完善了。過去流式處理是我們關(guān)注的重點，今年我們也開始逐漸的支持批處理，朝著批流一體的方向演進(jìn)。另外AI是目前比較火的一個方向，對于AI來說，它的實時化也是一個重要的研究方向。所以我們的實時計算平臺將會朝著批流一體和AI的方向進(jìn)行發(fā)展。
2.平臺架構(gòu)
上面是京東實時計算平臺JRC的整體架構(gòu)，整個架構(gòu)以定制化改造后的Flink為核心，F(xiàn)link運行在K8S上，狀態(tài)存儲在HDFS集群上，通過Zookeeper保證集群的高可用。支持流式源JDQ（京東基于Kafka深入定制實現(xiàn)的實時數(shù)據(jù)總線）和Hive，數(shù)據(jù)主要寫入JimDB（京東內(nèi)存數(shù)據(jù)庫）、ES、Hbase和京東OLAP。計算平臺支持SQL和普通JAR包兩種方式的作業(yè)，具有配置、部署、調(diào)試、監(jiān)控、和日志處理等功能。
3. 業(yè)務(wù)場景
京東Flink服務(wù)于京東內(nèi)部非常多的業(yè)務(wù)線，有70多個一級部門在使用，主要應(yīng)用場景包括實時數(shù)倉，實時大屏，實時推薦，實時報表，實時風(fēng)控和實時監(jiān)控，當(dāng)然還有其他一些應(yīng)用場景。對數(shù)據(jù)計算實時性有一定要求的場景，一般都會使用Flink進(jìn)行開發(fā)。
4. 業(yè)務(wù)規(guī)模
1.容器化歷程
京東從2018年開始進(jìn)行計算引擎的容器化改造，2019年初已經(jīng)實現(xiàn)計算單元全部容器化，2020年進(jìn)行了容器化方案升級，使用native k8s實現(xiàn)計算資源的彈性擴容。容器化改造的好處是提升了資源使用率，提高了研發(fā)效率，增強了業(yè)務(wù)穩(wěn)定性，減少了運維部署成本。
2.容器化方案
舊的容器化方案是基于k8s Deployment部署的Standalone Session集群，它需要事先預(yù)估出集群所需資源，比如需要的JobManager和TaskManager的資源規(guī)格和個數(shù)。然后JRC平臺通過K8S客戶端向K8S Master提出請求，創(chuàng)建JobManager的Deployment和TaskManager的Deployment。其中使用ZK保證高可用，使用Hdfs實現(xiàn)狀態(tài)存儲，使用Prometheus實現(xiàn)監(jiān)控指標(biāo)的上傳，結(jié)合Grafana實現(xiàn)指標(biāo)的直觀展示。集群使用ES存儲日志，方便日志的查詢。
3.容器化遇到的問題&對策
容器化過程中可能遇到很多問題：
①?JM/TM故障自動恢復(fù)
應(yīng)用部署在容器中，當(dāng)應(yīng)用出現(xiàn)異常時，如何發(fā)現(xiàn)應(yīng)用或者異常的情況呢？比如可以使用存活探針，編寫檢測腳本定期讀取應(yīng)用的心跳信息。當(dāng)檢測到Pod處于不健康狀態(tài)時，可以采用k8s的重啟機制來重啟不健康的容器。
② 減少Pod異常對業(yè)務(wù)影響
在k8s中由于硬件異常、資源過載、Pod不健康等問題會導(dǎo)致Pod被驅(qū)逐或自動重啟，Pod重啟時勢必會影響到該Pod上分布計算任務(wù)的正常運行。這個時候可以考慮采用適當(dāng)?shù)闹貑⒉呗?、改造?nèi)核等方案來減少對任務(wù)影響。比如京東實現(xiàn)了JM Failover優(yōu)化，當(dāng)Pod異常引起JM Failover時采用的是任務(wù)不恢復(fù)、重建任務(wù)狀態(tài)恢復(fù)的方式，可以一定程度上減少Pod重啟對業(yè)務(wù)帶來的影響。
③ 性能問題
在容器環(huán)境下，JVM對cpu和內(nèi)存的感知會有一定的問題，在Java8版本中，一些參數(shù)就要進(jìn)行顯式的設(shè)置。對于機器性能差異或熱點等問題導(dǎo)致部分Pod計算慢的問題，可以考慮進(jìn)行針對性優(yōu)化（比如實現(xiàn)基于負(fù)載的數(shù)據(jù)分發(fā)）或處理（比如檢測到計算慢的Pod將其驅(qū)逐到負(fù)載較低的機器）。此外，對于使用容器網(wǎng)絡(luò)的情況下，可能會帶來一定的網(wǎng)絡(luò)性能損耗，此時可以根據(jù)情況選擇使用主機網(wǎng)絡(luò)避免網(wǎng)絡(luò)虛擬化帶來的開銷，或者選擇更高性能的網(wǎng)絡(luò)插件。
④ 重要業(yè)務(wù)穩(wěn)定性
如何保證業(yè)務(wù)的穩(wěn)定性是一個需要重點考慮的問題。除了保證系統(tǒng)各個環(huán)節(jié)的高可用外，還可以根據(jù)業(yè)務(wù)情況考慮使用其它合理的方案，例如業(yè)務(wù)分級管理，獨立資源池，多機房互備等。
4.容器化方案升級（Native k8s）
原有容器化方案存在一定的問題：
資源需要提前分配
無法實現(xiàn)資源彈性伸縮
極端場景下Pod不能正常拉起，影響任務(wù)恢復(fù)
重要業(yè)務(wù)穩(wěn)定性
容器化升級的解決方案是采用Native K8s的方式。由JRC平臺先向K8S Master發(fā)出請求，創(chuàng)建JobManager的Deployment；然后在用戶通過Rest服務(wù)提交任務(wù)后，由JobMaster通過JDResourceManager 向JRC平臺發(fā)出請求，然后JRC平臺向 K8s Master 動態(tài)申請資源去創(chuàng)建運行TaskManager 的Pod。
此處，通過引入JRC平臺與K8s交互，屏蔽了不同容器平臺的差異，解耦了鏡像與平臺集群配置&邏輯變化。另外，為了兼容原有Slot分配策略，在提交任務(wù)時會預(yù)估出任務(wù)所需資源并一次性申請，之后采用等待一定時間后進(jìn)行slot分配的方式達(dá)到兼容目的。
主要做了以下四個方面的優(yōu)化：
性能
穩(wěn)定性
易用性
功能擴展
下邊分幾個重要的點進(jìn)行講解：
1.預(yù)覽拓?fù)?/strong>
預(yù)覽拓?fù)渲饕菫榱私鉀Q業(yè)務(wù)的一些痛點：比如任務(wù)調(diào)優(yōu)繁瑣、SQL任務(wù)無法指定并行度、任務(wù)需要的額Slot數(shù)不清楚、并行度調(diào)整后網(wǎng)絡(luò)buffer不足等。在Flink任務(wù)調(diào)試階段，對任務(wù)并行度、Slot分組、Chaining策略的調(diào)整是個反復(fù)的過程，如果把參數(shù)寫到命令行就太繁瑣了。而基于預(yù)覽拓?fù)渚涂梢院芊奖愕貙@些參數(shù)進(jìn)行配置。
預(yù)覽拓?fù)浠镜膶崿F(xiàn)方案如上圖：用戶提交JAR包后可根據(jù)JAR包生成對應(yīng)的拓?fù)鋱D，之后用戶根據(jù)拓?fù)鋱D可以進(jìn)行在線調(diào)整，最后自動將修改后的配置和原來的JAR包一起進(jìn)行任務(wù)提交。
預(yù)覽拓?fù)錂C制使得不修改程序多次提交任務(wù)調(diào)優(yōu)成為可能，但是如何保證前后兩次提交生成算子穩(wěn)定的對應(yīng)關(guān)系呢？解決方案的關(guān)鍵是保證算子有穩(wěn)定的唯一身份標(biāo)識，具體算法是：如果算子指定了uidHash就用uidHash，如果算子指定了uid就使用uid，否則就從source開始廣度優(yōu)先遍歷，利用算子在graph中的位置生成一個穩(wěn)定hash值。
2.背壓量化
第二個重要的優(yōu)化是背壓量化。
在Flink開發(fā)的時候，主要有兩種方式：
①通過Flink UI背壓面板觀察是否背壓。使用這種方式在某些場景比較方便，但是它存在幾個問題：
在有些場景下采集不到背壓
對于歷史背壓情況無法跟蹤
背壓影響不直觀
大并行度時背壓采集壓力
②通過任務(wù)背壓相關(guān)指標(biāo)進(jìn)行觀察和分析，通過將指標(biāo)定期采集并存儲起來，可以進(jìn)行實時或歷史的背壓分析。但是它也有一些不足的地方：
不同F(xiàn)link版本中指標(biāo)含義有一定差異
分析背壓有一定門檻，需要對于指標(biāo)含義有深入理解，聯(lián)合進(jìn)行分析
背壓未量化，對業(yè)務(wù)影響程度不夠直觀
京東的解決方案是采集背壓發(fā)生的位置、時間和次數(shù)指標(biāo)，并對這些指標(biāo)進(jìn)行上報存儲。同時對量化的背壓指標(biāo)結(jié)合運行時拓?fù)洌梢跃_反映發(fā)生背壓現(xiàn)場的情況。
3.HDFS優(yōu)化
隨著業(yè)務(wù)數(shù)量的增多，HDFS集群的壓力就會變得很大。這會直接導(dǎo)致RPC響應(yīng)時間變慢，造成請求堆積，同時大量小文件也會對NN內(nèi)存造成很大壓力。對此京東嘗試的解決方案有4方面：限制checkpoint最小間隔，時間最小設(shè)置在1min左右可以滿足大部分業(yè)務(wù)需求；進(jìn)行小文件合并；降低cp創(chuàng)建和刪除時的hdfs rpc請求；HDFS集群多ns分散均衡壓力。
4.網(wǎng)絡(luò)分發(fā)優(yōu)化
在實踐過程中我們發(fā)現(xiàn)，即使業(yè)務(wù)使用了rebalance并且對任務(wù)進(jìn)行了打散分布，但是由于機器處理能力和負(fù)載的差異，會導(dǎo)致任務(wù)各個并行度不同程序的背壓表現(xiàn)，嚴(yán)重影響了任務(wù)的性能。為此，我們開發(fā)了基于負(fù)載的動態(tài)rebalance，在數(shù)據(jù)進(jìn)行分發(fā)時優(yōu)先選擇下游負(fù)載最小的channel進(jìn)行分發(fā)。
經(jīng)測試，在特定場景下性能能夠提升近一倍。
5.ZK防抖
任務(wù)重啟的原因是由于在這些場景發(fā)生時，Curator會將狀態(tài)設(shè)置為suspended，并且Curator認(rèn)為suspended為Error狀態(tài)，從而會釋放leader，F(xiàn)link發(fā)現(xiàn)notleader后會revokeLeadership，從而造成任務(wù)重啟。
一個可行的解決辦法是升級Curator的版本，同時將connectionStateErrorPolicy設(shè)置為SessionConnetionStateErrorPolicy。
6.日志分離
目前我們一個集群是支持跑多個任務(wù)的，這時日志會出現(xiàn)的問題是：任務(wù)的日志和集群Framework日志混在一起，同時集群的多個任務(wù)日志也是混在一起的，不太方便用戶查看日志，快速定位問題。
為了解決這個問題，首先要弄清楚目前Flink加載日志框架的基本機制：為了避免跟業(yè)務(wù)Job中可能包含的日志框架的依賴、配置文件產(chǎn)生沖突，F(xiàn)link日志相關(guān)類的加載都代理給TaskManager框架的類加載器，也就是Parent Classloader，而框架加載的這些類都是從Flink安裝包的lib目錄下加載的。對于日志配置文件，F(xiàn)link通過 JVM 啟動參數(shù)來指定配置日志配置文件路徑。
日志分離的解決方案是：將日志相關(guān)jar包加入到各個task自己classloader（user classloader）的類路徑中；同時確保使用user classloader加載日志類和加載自己的日志配置；
另外對于使用了Flink框架的類（比如PrintSinkFunction），日志不能做到很好的分離，可以考慮使用logback MDC機制。
未來規(guī)劃主要包括四個方面:
① 統(tǒng)一計算引擎
引擎Storm全部升級為Flink，這樣可以減少平臺的運維成本，同時可以提高作業(yè)性能（目前已經(jīng)接近完成）。
② 更多SQL作業(yè)
持續(xù)完善SQL平臺，降低用戶的使用門檻，推動用戶更多使用SQL開發(fā)作業(yè)。
③ 智能運維
使用智能診斷，自適應(yīng)調(diào)整運行參數(shù)，提升任務(wù)的魯棒性
④ 批流一體
深度打造批流一體實時計算平臺，兼具低延遲的流處理和高性能的批處理能力。另外統(tǒng)一架構(gòu)，實現(xiàn)代碼復(fù)用，降低用戶的使用成本。
今天的分享就到這里，謝謝大家。
特別推薦一個分享架構(gòu)+算法的優(yōu)質(zhì)內(nèi)容，還沒關(guān)注的小伙伴，可以長按關(guān)注一下：
長按訂閱更多精彩▼
如有收獲，點個在看，誠摯感謝
免責(zé)聲明：本文內(nèi)容由21ic獲得授權(quán)后發(fā)布，版權(quán)歸原作者所有，本平臺僅提供信息存儲服務(wù)。文章僅代表作者個人觀點，不代表本平臺立場，如有問題，請聯(lián)系我們，謝謝！
下載文檔
