Flink在美團的實踐與應用

做者: 劉迪珊

本文整理自8月11日在北京舉行的Flink Meetup，分享嘉賓劉迪珊(2015年加入美團數據平臺。致力於打造高效、易用的實時計算平臺，探索不一樣場景下實時應用的企業級解決方案及統⼀化服務)。

美團實時計算平臺現狀和背景

實時平臺架構

上圖呈現的是當前美團實時計算平臺的簡要架構。最底層是數據緩存層，能夠看到美團測的全部日誌類的數據，都是經過統一的日誌收集系統收集到Kafka。Kafka做爲最大的數據中轉層，支撐了美團線上的大量業務，包括離線拉取，以及部分實時處理業務等。在數據緩存層之上，是一個引擎層，這一層的左側是咱們目前提供的實時計算引擎，包括Storm和Flink。Storm在此以前是 standalone 模式的部署方式，Flink因爲其如今運行的環境，美團選擇的是On YARN模式，除了計算引擎以外，咱們還提供一些實時存儲功能，用於存儲計算的中間狀態、計算的結果、以及維度數據等，目前這一類存儲包含Hbase、Redis以及ES。在計算引擎之上，是趨於五花八門的一層，這一層主要面向數據開發的同窗。實時數據開發面臨諸多問題，例如在程序的調試調優方面就要比普通的程序開發困難不少。在數據平臺這一層，美團面向用戶提供的實時計算平臺，不只能夠託管做業，還能夠實現調優診斷以及監控報警，此外還有實時數據的檢索以及權限管理等功能。除了提供面向數據開發同窗的實時計算平臺，美團如今正在作的事情還包括構建元數據中心。這也是將來咱們想作SQL的一個前提，元數據中心是承載實時流系統的一個重要環節，咱們能夠把它理解爲實時系統中的大腦，它能夠存儲數據的Schema，Meta。架構的最頂層就是咱們如今實時計算平臺支撐的業務，不只包含線上業務日誌的實時查詢和檢索，還涵蓋當下十分熱門的實時機器學習。機器學習常常會涉及到搜索和推薦場景，這兩個場景最顯著特色：1、會產生海量實時數據；2、流量的QPS至關高。此時就須要實時計算平臺承載部分實時特徵的提取工做，實現應用的搜索推薦服務。還有一類是比較常見的場景，包括實時的特徵聚合，斑馬Watcher（能夠認爲是一個監控類的服務），實時數倉等。

以上就是美團目前實時計算平臺的簡要架構。

實時平臺現狀

美團實時計算平臺的現狀是做業量如今已經達到了近萬，集羣的節點的規模是千級別的，天級消息量已經達到了萬億級，高峯期的消息量可以達到千萬條每秒。

痛點和問題

美團在調研使用Flink以前遇到了一些痛點和問題:

• 實時計算精確性問題：在調研使用Flink以前美團很大規模的做業是基於Storm去開發的，Storm主要的計算語義是At-Least-Once，這種語義在保證正確性上其實是有一些問題的，在Trident以前Storm是無狀態的處理。雖然Storm Trident提供了一個維護狀態的精確的開發，可是它是基於串行的Batch提交的，那麼遇到問題在處理性能上可能會有一點瓶頸。而且Trident是基於微批的處理，在延遲上沒有達到比較高的要求，因此不能知足一些對延遲比較高需求的業務。
• 流處理中的狀態管理問題：基於以前的流處理過程當中狀態管理的問題是很是大的一類問題。狀態管理除了會影響到好比說計算狀態的一致性，還會影響到實時計算處理的性能以及故障恢復時候的能力。而Flink最突出的一個優點就是狀態管理。
• 實時計算表義能力的侷限性：在實時計算以前不少公司大部分的數據開發仍是面向離線的場景，近幾年實時的場景也慢慢火熱起來了。那與離線的處理不一樣的是，實時的場景下，數據處理的表意能力可能有必定的限制，好比說他要進行精確計算以及時間窗口都是須要在此之上去開發不少功能性的東西。
• 開發調試成本高：近千結點的集羣上已經跑了近萬的做業，分佈式的處理的引擎，手工寫代碼的方式，給數據開發的同窗也帶來了很高開發和調試的成本，再去維護的時候，運維成本也比較高。

Flink探索關注點

在上面這些痛點和問題的背景下，美團從去年開始進行Flink的探索，關注點主要有如下4個方面：

• ExactlyOnce計算能力
• 狀態管理能力
• 窗口/Join/時間處理等等
• SQL/TableAPI

Flink在美團的實踐

下面帶你們來看一下，美團從去年投入生產過程當中都遇到了哪些問題，以及一些解決方案，分爲下面三個部分：

穩定性實踐

穩定性實踐-資源隔離

1.資源隔離的考慮：分場景、按業務

• 高峯期不一樣，運維時間不一樣；
• 可靠性、延遲需求不一樣；
• 應用場景，重要性不一樣；

  2.資源隔離的策略：

• YARN打標籤，節點物理隔離；
• 離線DataNode與實時計算節點的隔離；

穩定性實踐-智能調度

智能調度目的也是爲了解決資源不均的問題，如今普通的調度策略就是基於CPU，基於內存去調度的。除此以外，在生產過程當中也發現了一些其餘的問題，好比說Flink是會依賴本地磁盤，進行依賴本地磁盤作本地的狀態的存儲，因此磁盤IO，還有磁盤的容量，也是一類考慮的問題點，除此以外還包括網卡流量，由於每一個業務的流量的狀態是不同的，分配進來會致使流量的高峯，把某一個網卡打滿，從而影響其餘業務，因此指望的話是說作一些智能調度化的事情。目前暫時能作到的是從cpu和內存兩方面，將來會從其餘方面作一些更優的調度策略。

穩定性實踐-故障容錯

1.節點/網絡故障

• JobManagerHA
• 自動拉起

與Storm不一樣的是，知道Storm在遇到異常的時候是很是簡單粗暴的，好比說有發生了異常，可能用戶沒有在代碼中進行比較規範的異常處理，可是沒有關係，由於worker會重啓做業還會繼續執行，而且他保證的是At-Least-Once這樣的語義，好比說一個網絡超時的異常對他而言影響可能並無那麼大，可是Flink不一樣的是他對異常的容忍度是很是的苛刻的，那時候就考慮的是好比說會發生節點或者是網絡的故障，那JobManager單點問題可能就是一個瓶頸，JobManager那個若是掛掉的話，那麼可能對整個做業的影響就是不可回覆的，因此考慮了作HA，另一個就是會去考慮一些因爲運維的因素而致使的那做業，還有除此以外，可能有一些用戶做業是沒有開啓CheckPoint，但若是是由於節點或者是網絡故障致使掛掉，但願會在平臺內層作一些自動拉起的策略，去保證做業運行的穩定性。

2.上下游容錯

• FlinkKafka08異常重試

咱們的數據源主要是Kafka，讀寫Kafka是一類很是常見的實時流處理避不開的一個內容，而Kafka自己的集羣規模是很是比較大的，所以節點的故障出現是一個常態問題，在此基礎上咱們對節點故障進行了一些容錯，好比說節點掛掉或者是數據均衡的時候，Leader會切換，那自己Flink的讀寫對Leader的切換容忍度沒有那麼高，在此基礎上咱們對一些特定場景的，以及一些特有的異常作的一些優化，進行了一些重試。

3.容災

• 多機房
• 流熱備

容災可能你們對考慮的並很少，好比說有沒有可能一個機房的全部的節點都掛掉了，或者是沒法訪問了，雖然它是一個小几率的事件，但它也是會發生的。因此如今也會考慮作多機房的一些部署，包括還有Kafka的一些熱備。

Flink平臺化

Flink平臺化-做業管理

在實踐過程當中，爲了解決做業管理的一些問題，減小用戶開發的一些成本，咱們作了一些平臺化的工做，下圖是一個做業提交的界面展現，包括做業的配置，做業生命週期的管理，報警的一些配置，延遲的展現，都是集成在實時計算平臺的。

Flink平臺化-監控報警

在監控上咱們也作了一些事情，對於實時做業來說，對監控的要求會更高，好比說在做業延遲的時候對業務的影響也比較大，因此作了一些延遲的報警，包括做業狀態的報警，好比說做業存活的狀態，以及做業運行的狀態，還有將來會作一些自定義Metrics的報警。自定義Metrics是將來會考慮基於做業處理自己的內容性，作一些可配置化的一些報警。

Flink平臺化-調優診斷

• 實時計算引擎提供統一日誌和Metrics方案
• 爲業務提供按條件過濾的日誌檢索
• 爲業務提供自定義時間跨度的指標查詢
• 基於日誌和指標，爲業務提供可配置的報警

另外就是剛剛提到說在開發實時做業的時候，調優和診斷是一個比較難的痛點，就是用戶不是很難去查看分佈式的日誌，因此也提供了一套統一的解決方案。這套解決方案主要是針對日誌和Metrics，會在針對引擎那一層作一些日誌和Metrics的上報，那麼它會經過統一的日誌收集系統，將這些原始的日誌，還有Metrics聚集到Kafka那一層。從此Kafka這一層你們能夠發現它有兩個下游，一方面是作日誌到ES的數據同步，目的的話是說可以進入日誌中心去作一些日誌的檢索，另一方面是經過一些聚合處理流轉到寫入到OpenTSDB把數據作依賴，這份聚合後的數據會作一些查詢，一方面是Metrics的查詢展現，另一方面就是包括實作的一些相關的報警。

下圖是當前某一個做業的一個可支持跨天維度的Metrics的一個查詢的頁面。能夠看到說若是是可以經過縱向的對比，能夠發現除了做業在某一個時間點是由於什麼狀況致使的？好比說延遲啊這樣容易幫用戶判斷一些他的作做業的一些問題。除了做業的運行狀態以外，也會先就是採集一些節點的基本信息做爲橫向的對比

下圖是當前的日誌的一些查詢，它記錄了，由於做業在掛掉以後，每個ApplicationID可能會變化，那麼基於做業惟一的惟一的主鍵做業名去搜集了全部的做業，從建立之初到當前運行的日誌，那麼能夠容許用戶的跨Application的日誌查詢。

生態建設

爲了適配這兩類MQ作了不一樣的事情，對於線上的MQ，指望去作一次同步屢次消費，目的是避免對線上的業務形成影響，對於的生產類的Kafka就是線下的Kafka，作了一些地址的地址的屏蔽，還有基礎基礎的一些配置，包括一些權限的管理，還有指標的採集。

Flink在美團的應用

下面會給你們講兩個Flink在美團的真實使用的案例。第一個是Petra，Petra實際上是一個實時指標的一個聚合的系統，它實際上是面向公司的一個統一化的解決方案。它主要面向的業務場景就是基於業務的時間去統計，還有計算一些實時的指標，要求的話是低時延，他還有一個就是說，由於它是面向的是通用的業務，因爲業務多是各自會有各自不一樣的維度，每個業務可能包含了包括應用通道機房，還有其餘的各自應用各個業務特有的一些維度，並且這些維度可能涉及到比較多，另一個就是說它多是就是業務須要去作一些複合的指標的計算，好比說最多見的交易成功率，他可能須要去計算支付的成功數，還有和下單數的比例。另一個就是說統一化的指標聚合可能面向的仍是一個系統，好比說是一些B端或者是R段的一些監控類的系統，那麼系統對於指標系統的訴求，就是說我但願指標聚合可以最真最實時最精確的可以產生一些結果，數據保證說它的下游系統可以真實的監控到當前的信息。右邊圖是我當一個Metrics展現的一個事例。能夠看到其餘其實跟剛剛講也是比較相似的，就是說包含了業務的不一樣維度的一些指標匯聚的結果。

Petra實時指標聚合

1.業務場景：

• 基於業務時間（事件時間）
• 多業務維度：如應用、通道、機房等
• 複合指標計算：如交易成功率=支付成功數/下單數
• 低延遲：秒級結果輸出

2.Exactlyonce的精確性保障

• Flinkcheckpoint機制

3.維度計算中數據傾斜

• 熱點key散列

4.對晚到數據的容忍能力

• 窗口的設置與資源的權衡

在用Flink去作實時指標複覈的系統的時候，着重從這幾方面去考慮了。第一個方面是說精確的計算，包括使用了FLink和CheckPoint的機制去保證說我能作到不丟不重的計算，第一個首先是由統一化的Metrics流入到一個預聚合的模塊，預聚合的模塊主要去作一些初始化的一些聚合，其中的爲何會分預聚合和全量聚合主要的解決一類問題，包括就剛剛那位同窗問的一個問題，就是數據傾斜的問題，好比說在熱點K發生的時候，當前的解決方案也是經過預聚合的方式去作一些緩衝，讓儘可能把K去打散，再聚合全量聚合模塊去作匯聚。那其實也是隻能解決一部分問題，因此後面也考慮說在性能的優化上包括去探索狀態存儲的性能。下面的話仍是包含晚到數據的容忍能力，由於指標匯聚可能剛剛也提到說要包含一些複合的指標，那麼符合的指標所依賴的數據可能來自於不一樣的流，即使來自於同一個流，可能每個數據上報的時候，可能也會有晚到的狀況發生，那時候須要去對數據關聯作晚到的容忍，容忍的一方面是說能夠設置晚到的Lateness的延遲，另外一方面是能夠設置窗口的長度，可是其實在現實的應用場景上，其實還有一方面考慮就是說除了去儘可能的去拉長時間，還要考慮真正的計算成本，因此在這方面也作了一些權衡，那麼指標基本就是通過全量聚合以後，聚合結果會回寫Kafka，通過數據同步的模塊寫到OpenTSDB去作，最後去grafana那作指標的展現，另外一方面可能去應用到經過Facebook包同步的模塊去同步到報警的系統裏面去作一些指標，基於指標的報警。

下圖是如今提供的產品化的Petra的一個展現的機示意圖，能夠看到目前的話就是定義了某一些經常使用的算子，以及維度的配置，容許用戶進行配置話的處理，直接去可以獲取到他指望要的指標的一個展現和匯聚的結果。目前還在探索說爲Petra基於Sql作一些事情，由於不少用戶也比較就是在就是習慣上也能夠傾向於說我要去寫Sql去完成這樣的統計，因此也會基於此說依賴Flink的自己的對SQl還有TableAPI的支持，也會在Sql的場景上進行一些探索。

MLX機器學習平臺

第二類應用就是機器學習的一個場景，機器學習的場景可能會依賴離線的特徵數據以及實時的特徵數據。一個是基於現有的離線場景下的特徵提取，通過了批處理，流轉到了離線的集羣。另一個就是近線模式，近線模式出的數據就是現有的從日誌收集系統流轉過來的統一的日誌，通過Flink的處理，就是包括流的關聯以及特徵的提取，再作模型的訓練，流轉到最終的訓練的集羣，訓練的集羣會產出P的特徵，還有都是Delta的特徵，最終將這些特徵影響到線上的線上的特徵的一個訓練的一個服務上。這是一個比較常見的，好比說比較就是通用的也是比較通用的一個場景，目前的話主要應用的方可能包含了搜索還有推薦，以及一些其餘的業務。

將來展望

將來的話可能也是經過也是指望在這三方面進行作一些更多的事情，剛剛也提到了包括狀態的管理，第一個是狀態的統一的，好比說Sql化的統一的管理，但願有統一的配置，幫用戶去選擇一些指望的回滾點。另一個就是大狀態的性能優化，由於好比說像作一些流量數據的雙流的關聯的時候，如今也遇到了一些性能瓶頸的問題，對於說啊基於內存型的狀態，基於內存型的數據的處理，以及基於RocksDB的狀態的處理，作過性能的比較，發現其實性能的差別仍是有一些大的，因此但願說在基於RocksDBBackend的上面可以去儘可能去更多的作一些優化，從而提高做業處理的性能。第二方面就是Sql，Sql的話應該是每個位就是當前可能各個公司都在作的一個方向，由於以前也有對Sql作一些探索，包括提供了基於Storm的一些Sql的表示，可是可能對於以前的話對於與語義的表達可能會有一些欠缺，因此但願說在基於Flink可去解決這些方面的事情，以及包括Sql的併發度的一些配置的優化，包括Sql的查詢的一些優化，都但願說在Flink將來可以去優化更多的東西，去真正能使Sql應用到生產的環境。

另一方面的話就是會進行新的場景的也在作新的場景的一些探索，指望是好比說包括剛剛也提到說除了流式的處理，也指望說把離線的場景下的數據進行一些合併，經過統一的Sql的API去提供給業務作更多的服務，包括流處理，還有批處理的結合。

更多資訊請訪問 Apache Flink 中文社區網站