馬蜂窩實時計算平臺演進之路

MES 是馬蜂窩統一實時計算平臺，爲各條業務線提供穩定、高效的實時數據計算和查詢服務。在總體設計方面，MES 借鑑了 Lambda 架構的思想。本篇文章，咱們將從四個方面瞭解 MES：

1. 關於 Lambda 架構

2.MES 架構和原理

3.MES 優化歷程

4. 近期規劃

 

關於 Lambda 架構

Lambda 架構是由 Storm 做者 NathanMarz 根據本身在 Twitter 的分佈式數據處理系統經驗，提出的一個實時大數據處理框架，具備高容錯、低延時和可擴展等特性。

Lambda 架構核心的思想主要能夠概括成兩點：

（1）數據從上游 MQ 消息中間件過來後分爲 2 路，一路離線批處理, 一路實時處理並有各自的 View 以供查詢。

（2）Query 時，對數據作 Function, 結合 Batch View 和 Realtime View，獲得最終結果。

具體來講，Lambda 架構將大數據系統架構爲多個層次：批處理層（Batch layer）、實時處理層（Speed Layer）、服務層（Serving Layer）。

咱們結合一張經典的 Lambda 架構圖分別來看：

圖 1：Lambda 架構

（圖片來源：網絡）

批處理層（Batch Layer）：批處理層承擔的任務是對從上游進來的全部被系統認爲不可變的原始數據。類比目前的數據平臺架構來看, 即離線的那幾張保存原始數據的主表。這 3 張主表是全部完整的數據而且是不可變的，基於這幾張主表，數據通過 Batch 、ETL，產生供批處理查詢的 Batch View。

加速層（Speed Layer）：批處理層雖然能夠很好地處理離線數據，但它不能很好知足對於時間粒度的需求。對於須要不斷實時生成和實時查詢處理的數據，一般會放在加速層來進行實時處理和轉化。

加速層與批處理層最大的區別在於，加速層只處理最近的數據，而批處理層處理全部數據。另外在數據的讀取方面，爲了知足最小延遲，加速層不會在同一數據讀取全部新數據，而是在收到新數據時更新 Realtime View，因此咱們說，在加速層進行的是一種增量的計算。 

服務層（Serving Layer）：服務層用於響應用戶的查詢請求，合併 Batch View 和 Realtime View 中的結果數據集到最終的數據集，並向外對用戶經過統一接口，提供實時+離線的數據統計服務。

基於 Lambda 的數據平臺架構, 能夠按照分層集成衆多的大數據組件。在對 MES 的架構設計中，咱們借鑑了 Lambda 架構的思想來實現更快

更準、魯棒性更好的特性。 

 

馬蜂窩實時計算平臺 MES

爲了保證 MES 實時計算平臺的性能，咱們結合馬蜂窩的實際業務場景，主要圍繞低延遲，高吞吐、容災能力和 Exacty Once 的流式語義這四點，來進行架構設計和技術選型。

總體架構設計

對照 Lambda 架構，咱們選用 Kafka 做爲消息中間件，批處理層選擇 Hive、Presto，加速層也就是實時處理層選擇 Spark、Flink 等。

圖 2：MES 總體架構圖

數據從 Kafka 出來後走兩條線，一條是 Spark Streaming，支持秒級別的實時數據，計算結果會入庫到 Redis 裏。次日凌晨，Redis 中前一天的全部數據 Batch 到 HBase 中；

另一條是 Flink+Druid，用來處理分鐘級和小時級的數據；

上面提供一層 Restful API / Thrift API 封裝，供 MES 頁面或其餘業務經過接口的方式來獲取數據；

若是實時數據出了問題，咱們會經過 HDFS 中的離線主表進行重算，也是有兩條路徑：

一是爲用戶服務的 MES 重算系統，用戶能夠自助化選取重算規則，提交重算任務。這個任務會被提交到 PrestoSQL 集羣，計算結果最終落地到 HBase 裏，重算後 MES 的歷史數據就會和離線數據算出來的數據保持一致；

另一條線是 Spark 全量重算，由數據平臺的小夥伴內部使用，解決的是基於全部事件組、全部規則的全天數據重算。Spark 會讀取配置規則，重算全部前一天的數據後入庫到 HBase，保持實時數據和離線數據的一致性；

監控系統是 Grafana，它開放了通用接口給 Python、Java 等語言來上報相關信息，只要按照接口上報要想關注的指標並進行簡單配置，就能夠查詢結果，好比 MES 的延遲時間、一些 Restful 接口的調用量等, 若是出現不正常的狀況將經過郵件告警；

最右邊是貫穿始終的 MES 規則，咱們能夠抽象地把它看做是實時的配置流。

 

MES 實時計算引擎

1. 技術選型

結合馬蜂窩的業務需求，咱們對三大主流實時計算引擎 Storm、Spark Streaming、Flink 進行了選型對比。

Storm

Storm 是第一代流式計算引擎，實現了一個數據流 (Data Flow) 的模型。咱們能夠把它想象成一個發射點，一條一條產生數據，造成的數據流分佈式地在集羣上按照 Bolt 的計算邏輯進行轉換，完成計算、過濾等操做，在下游實現聚合。

Storm 的優點是實時性好，能夠達到毫秒級。可是它的吞吐量欠佳，而且只能爲消息提供「至少一次」的處理機制, 這意味着能夠保證每條消息都能被處理，但也可能發生重複。

Spark Streaming

Spark Streaming 不像 Storm 那樣一次一個地處理數據流，而是在處理前按時間間隔預先將其切分爲一段一段，進行「微批次」處理做業。這樣一來解決了吞吐量的問題，但它的實時性就沒有 Storm 那麼高，不過也能夠達到秒級處理。

在流式語義方面，因爲 Spark Streaming 容錯機制基於 RDD，依靠 CheckPoint，出錯以後會從該位置從新計算，不會致使重複計算。固然咱們也能夠本身來管理 offset，保證 Exactly Once (只算一次的語義) 的處理。

Flink

Flink 是新一代流式計算引擎，國內的阿里就是 Flink 的重度使用和貢獻者。Flink 是原生的流處理系統，把全部的數據都當作是流，認爲批處理是流處理中的一種特殊狀況。數據基於 Flink Stream Source 流入，中間通過 Operator，從 Sink 流出。

爲了解決流處理的容錯問題，Flink 巧妙地運用了分佈式快照的設計與可部分重發的數據源實現容錯。用戶可自定義對整個 Job 進行快照的時間間隔。當任務失敗時，Flink 會將整個 Job 恢復到最近一次快照，並從數據源重發快照以後的數據。Flink 同時保證了實時性和吞吐量，流式語義也作得很是好，可以保證 Exactly Once。

在此以外，組件技術選型的時候在知足本身業務現狀的同時, 還須要從之前幾個方面考慮:

• 開源組件是否能覆蓋需求
• 開源組件的擴展性和二次開發的難度
• 開源組件 API 是否穩定
• 開源組件是否有應用於生產環境的案例，好比多少公司應用於生產環境
• 開源組件社區是否活躍，好比能夠看 github，issues，jiras 這些活躍程度
• 開源組件 License 限定問題
• 開源組件之間的耦合問題

2. 設計

下圖描述了 MES 實時計算引擎處理數據的過程：

圖 3：MES Streaming

數據從 Kafka 源源不斷地過來造成數據流，用戶經過 UI 配置的一些規則造成實時配置流，數據流和配置流進入到實時計算引擎 Spark Streaming 後進行聚合計算。計算出的實時數據寫入到 Redis，歷史數據入庫到 HBase。UI 目前經過 Restful API 來獲取實時和歷史數據。

3. 演進

關於 MES 實時計算的引擎，咱們主要經歷了兩次演進。

第一代 ：Spark Streaming + Redis + HBase 

在設計第一代 MES 時，咱們但願能夠支持秒級的計算，而且精確計算每個用戶。因此在當時的現狀下，咱們綜合考慮選擇了 Spark Streaming。

這個方案計算出來的 UV 是比較精確的。但它有本身的侷限性：

首先，這一套架構用到的幾個組件其實對資源都比較依賴, 並且 SparkStreaming 對那種時不時的流量高峯的數據處理不是很是友好。數據先在 Spark Streaming 算好而後再入 Redis，最後再入庫到 Hbase，數據鏈路比較長，很差維護。

另外，第一代 MES 只支持自助配置規則，有了規則纔會實時計算。因此對於比較自由的 OLAP 交叉分析不友好。並且若是因爲集羣不穩定等緣由致使的任務失敗數據少算, 那麼不論是用戶自助提交 Presto 仍是利用 Spark 批處理全量重算，都是一個消耗集羣資源的過程。因爲批處理重算須要必定的時間來完成對歷史數據的修復，這對一些須要數據準確並及時提供的用戶不是很是友好。

咱們考慮，在數據量大的狀況下，咱們是否是能夠適當犧牲 UV 精準度的計算，來保障整個系統的性能和穩定性。因此就引入了 Flink +  Druid。

第二代：引入 Flink + Druid

剛纔咱們已經簡單瞭解過 Flink，再來講下 Druid。

Druid 是一個大數據實時查詢和分析的高容錯、高性能的開源分佈式系統，用來快速處理大規模的數據，它可以實現對大量數據的快速查詢和分析，不足是存在一個 2% 的偏差。但事實上，在數據量很是大的狀況下，2% 的偏差是能夠接受的。後面咱們又經過 Yahoo 提供的 Data Sketch，實現 UV 計算的精確調控，能夠實如今 800w 之下的數據量，UV 都是準確的。最終的計算結果經過 restful 接口提供給 MES 獲取數據並展示。

圖 4：關於 Druid

Flink + Druid 部分主要是用來處理數據量大、維度多，且不須要精確到秒級的業務數據，好比 Page logdata、mobile page、以及 Server Push。在最近 15 天的數據是能夠精確到分鐘級別查詢的，對於歷史數據，粒度越精確，持久化到 Druid 裏面的數據量就越大。

在離線批量導入部分，目前 Druid 支持小時級以及 T+1 天級的數據校訂。由於若是在 Flink +Tranquility 實時攝取數據這一階段 task 有異常的話，會致使實時數據到 Druid 有丟失的狀況出現。所以根據 Lambda 架構的思想，咱們能夠用小時級或者天級離線數據對丟失的數據進行重算補全。

對比一下兩代計算引擎，Flink + Druid 的引入很好地解決了由於大量數據的 UV 計算帶來的壓力：

圖 5：兩代實時計算引擎

MES 優化歷程

爲了更好地知足業務需求，提高整個系統的性能，咱們也在不斷對 MES 進行優化，主要包括實時計算集羣、計算引擎、查詢接口和監控方面。這裏主要和你們分享兩點。

1. 實時計算集羣優化

• Spark，Druid，Flink 集羣框架版本升級及相關參數優化；
• Redis，Hbase 節點擴容和參數優化;
• 集羣網絡，Yarn，Mesos 等資源管理框架調整和優化

2.實時計算引擎優化

數據結構和計算邏輯

對於 Spark 來說，Prefer 原生數據類型以及數組結構，對於指針類型以及嵌套的結構處理起來性能不是很是友好。所以要注意這一點，妥善優化本身的數據結構。

計算邏輯的部分也要考慮好。好比寫 Redis 的時候是事先規劃好要存入 Redis 中的數據結構來利用 Akka 併發每條來寫入，仍是在 Streaming 中算好一批結果最後來一次性寫入 Redis，這 2 種方式在性能上仍是有很大區別的。

參數優化

(1) 序列化方式首先是 Kyro 的方式，其次纔是 Java，序列化的方式不一樣對網絡的傳輸以及處理起來的性能是有影響的。

(2)Spark 推測執行機制。根據咱們集羣目前的現狀，有各類各樣的任務同時在跑，若是遇到集羣資源使用高峯期，那麼一個 Spark 任務落在比較慢的節點上就會拖累整個 Job 的執行進度。開啓推測執行以後，系統會把進程慢的任務主動殺死，而後從新產生一個相同的任務分配到資源充沛的節點上去快速完成它。

(3) 數據本地化。分佈式計算有一個經典的理念是：移動數據不如移動計算。好比說我把一個任務分紅不少並行的任務，有可能得到的任務恰好須要處理的數據就在處理的節點上，也有可能不是。因此這裏有一個本地化等待時間的參數能夠控制數據本地化的處理等級並對性能產生很大影響。

另外還用一些關於並行度控制、JVM GC 方面的調優就比較細節了，若是你們感興趣能夠留言給咱們交流。

 

將來規劃

馬蜂窩實時計算平臺的發展還須要不斷探索，將來咱們主要會在如下幾個方面重點推動：

1. 實時計算任務統一資源管理和任務調度

2. 支持複雜的實時 SQL OLAP 計算

3. 實時數據血緣關係及監控預警

4. 複雜實時 CEP 規則系統

感謝關注，歡迎你們掃描下方二維碼訂閱「馬蜂窩技術」內容並推薦給更多熱愛技術的朋友，但願有更多機會和你們交流。

微信關注「馬蜂窩技術」公衆號，閱讀更多技術乾貨。