MES 是馬蜂窩統(tǒng)一實(shí)時(shí)計(jì)算平臺(tái)，為各條業(yè)務(wù)線提供穩(wěn)定、高效的實(shí)時(shí)數(shù)據(jù)計(jì)算和查詢服務(wù)。在整體設(shè)計(jì)方面，MES 借鑒了 Lambda 架構(gòu)的思想。本篇文章，我們將從四個(gè)方面了解 MES：

1. 關(guān)于 Lambda 架構(gòu)

2.MES 架構(gòu)和原理

3.MES 優(yōu)化歷程

4. 近期規(guī)劃

關(guān)于 Lambda 架構(gòu)

Lambda 架構(gòu)是由 Storm 作者 NathanMarz 根據(jù)自己在 Twitter 的分布式數(shù)據(jù)處理系統(tǒng)經(jīng)驗(yàn)，提出的一個(gè)實(shí)時(shí)大數(shù)據(jù)處理框架，具有高容錯(cuò)、低延時(shí)和可擴(kuò)展等特性。

Lambda 架構(gòu)核心的思想主要可以歸納成兩點(diǎn)：

(1)數(shù)據(jù)從上游 MQ 消息中間件過(guò)來(lái)后分為 2 路，一路離線批處理, 一路實(shí)時(shí)處理并有各自的 View 以供查詢。

(2)Query 時(shí)，對(duì)數(shù)據(jù)做 Function, 結(jié)合 Batch View 和 Realtime View，得到最終結(jié)果。

具體來(lái)說(shuō)，Lambda 架構(gòu)將大數(shù)據(jù)系統(tǒng)架構(gòu)為多個(gè)層次：批處理層(Batch layer)、實(shí)時(shí)處理層(Speed Layer)、服務(wù)層(Serving Layer)。

我們結(jié)合一張經(jīng)典的 Lambda 架構(gòu)圖分別來(lái)看：

圖 1：Lambda 架構(gòu)

(圖片來(lái)源：網(wǎng)絡(luò))

批處理層(Batch Layer)：批處理層承擔(dān)的任務(wù)是對(duì)從上游進(jìn)來(lái)的所有被系統(tǒng)認(rèn)為不可變的原始數(shù)據(jù)。類比目前的數(shù)據(jù)平臺(tái)架構(gòu)來(lái)看, 即離線的那幾張保存原始數(shù)據(jù)的主表。這 3 張主表是所有完整的數(shù)據(jù)并且是不可變的，基于這幾張主表，數(shù)據(jù)經(jīng)過(guò) Batch 、ETL，產(chǎn)生供批處理查詢的 Batch View。

加速層(Speed Layer)：批處理層雖然可以很好地處理離線數(shù)據(jù)，但它不能很好滿足對(duì)于時(shí)間粒度的需求。對(duì)于需要不斷實(shí)時(shí)生成和實(shí)時(shí)查詢處理的數(shù)據(jù)，通常會(huì)放在加速層來(lái)進(jìn)行實(shí)時(shí)處理和轉(zhuǎn)化。

加速層與批處理層***的區(qū)別在于，加速層只處理最近的數(shù)據(jù)，而批處理層處理所有數(shù)據(jù)。另外在數(shù)據(jù)的讀取方面，為了滿足最小延遲，加速層不會(huì)在同一數(shù)據(jù)讀取所有新數(shù)據(jù)，而是在收到新數(shù)據(jù)時(shí)更新 Realtime View，所以我們說(shuō)，在加速層進(jìn)行的是一種增量的計(jì)算。

服務(wù)層(Serving Layer)：服務(wù)層用于響應(yīng)用戶的查詢請(qǐng)求，合并 Batch View 和 Realtime View 中的結(jié)果數(shù)據(jù)集到最終的數(shù)據(jù)集，并向外對(duì)用戶通過(guò)統(tǒng)一接口，提供實(shí)時(shí)+離線的數(shù)據(jù)統(tǒng)計(jì)服務(wù)。

基于 Lambda 的數(shù)據(jù)平臺(tái)架構(gòu), 可以按照分層集成眾多的大數(shù)據(jù)組件。在對(duì) MES 的架構(gòu)設(shè)計(jì)中，我們借鑒了 Lambda 架構(gòu)的思想來(lái)實(shí)現(xiàn)更快、更準(zhǔn)、魯棒性更好的特性。

馬蜂窩實(shí)時(shí)計(jì)算平臺(tái) MES

為了保證 MES 實(shí)時(shí)計(jì)算平臺(tái)的性能，我們結(jié)合馬蜂窩的實(shí)際業(yè)務(wù)場(chǎng)景，主要圍繞低延遲，高吞吐、容災(zāi)能力和 Exacty Once 的流式語(yǔ)義這四點(diǎn)，來(lái)進(jìn)行架構(gòu)設(shè)計(jì)和技術(shù)選型。

整體架構(gòu)設(shè)計(jì)

對(duì)照 Lambda 架構(gòu)，我們選用 Kafka 作為消息中間件，批處理層選擇 Hive、Presto，加速層也就是實(shí)時(shí)處理層選擇 Spark、Flink 等。

圖 2：MES 整體架構(gòu)圖

數(shù)據(jù)從 Kafka 出來(lái)后走兩條線，一條是 Spark Streaming，支持秒級(jí)別的實(shí)時(shí)數(shù)據(jù)，計(jì)算結(jié)果會(huì)入庫(kù)到 Redis 里。第二天凌晨，Redis 中前一天的所有數(shù)據(jù) Batch 到 HBase 中;

另外一條是 Flink+Druid，用來(lái)處理分鐘級(jí)和小時(shí)級(jí)的數(shù)據(jù);

上面提供一層 Restful API / Thrift API 封裝，供 MES 頁(yè)面或其他業(yè)務(wù)通過(guò)接口的方式來(lái)獲取數(shù)據(jù);

如果實(shí)時(shí)數(shù)據(jù)出了問(wèn)題，我們會(huì)通過(guò) HDFS 中的離線主表進(jìn)行重算，也是有兩條路徑：

• 一是為用戶服務(wù)的 MES 重算系統(tǒng)，用戶可以自助化選取重算規(guī)則，提交重算任務(wù)。這個(gè)任務(wù)會(huì)被提交到 PrestoSQL 集群，計(jì)算結(jié)果最終落地到 HBase 里，重算后 MES 的歷史數(shù)據(jù)就會(huì)和離線數(shù)據(jù)算出來(lái)的數(shù)據(jù)保持一致;
• 另外一條線是 Spark 全量重算，由數(shù)據(jù)平臺(tái)的小伙伴內(nèi)部使用，解決的是基于所有事件組、所有規(guī)則的全天數(shù)據(jù)重算。Spark 會(huì)讀取配置規(guī)則，重算所有前一天的數(shù)據(jù)后入庫(kù)到 HBase，保持實(shí)時(shí)數(shù)據(jù)和離線數(shù)據(jù)的一致性;

監(jiān)控系統(tǒng)是 Grafana，它開(kāi)放了通用接口給 Python、Java 等語(yǔ)言來(lái)上報(bào)相關(guān)信息，只要按照接口上報(bào)要想關(guān)注的指標(biāo)并進(jìn)行簡(jiǎn)單配置，就可以查詢結(jié)果，比如 MES 的延遲時(shí)間、一些 Restful 接口的調(diào)用量等, 如果出現(xiàn)不正常的情況將通過(guò)郵件告警;

最右邊是貫穿始終的 MES 規(guī)則，我們可以抽象地把它看作是實(shí)時(shí)的配置流。

MES 實(shí)時(shí)計(jì)算引擎

1. 技術(shù)選型

結(jié)合馬蜂窩的業(yè)務(wù)需求，我們對(duì)三大主流實(shí)時(shí)計(jì)算引擎 Storm、Spark Streaming、Flink 進(jìn)行了選型對(duì)比。

Storm

Storm 是***代流式計(jì)算引擎，實(shí)現(xiàn)了一個(gè)數(shù)據(jù)流 (Data Flow) 的模型。我們可以把它想象成一個(gè)發(fā)射點(diǎn)，一條一條產(chǎn)生數(shù)據(jù)，形成的數(shù)據(jù)流分布式地在集群上按照 Bolt 的計(jì)算邏輯進(jìn)行轉(zhuǎn)換，完成計(jì)算、過(guò)濾等操作，在下游實(shí)現(xiàn)聚合。

Storm 的優(yōu)勢(shì)是實(shí)時(shí)性好，可以達(dá)到毫秒級(jí)。但是它的吞吐量欠佳，并且只能為消息提供「至少一次」的處理機(jī)制, 這意味著可以保證每條消息都能被處理，但也可能發(fā)生重復(fù)。

Spark Streaming

Spark Streaming 不像 Storm 那樣一次一個(gè)地處理數(shù)據(jù)流，而是在處理前按時(shí)間間隔預(yù)先將其切分為一段一段，進(jìn)行「微批次」處理作業(yè)。這樣一來(lái)解決了吞吐量的問(wèn)題，但它的實(shí)時(shí)性就沒(méi)有 Storm 那么高，不過(guò)也可以達(dá)到秒級(jí)處理。

在流式語(yǔ)義方面，由于 Spark Streaming 容錯(cuò)機(jī)制基于 RDD，依靠 CheckPoint，出錯(cuò)之后會(huì)從該位置重新計(jì)算，不會(huì)導(dǎo)致重復(fù)計(jì)算。當(dāng)然我們也可以自己來(lái)管理 offset，保證 Exactly Once (只算一次的語(yǔ)義) 的處理。

Flink

Flink 是新一代流式計(jì)算引擎，國(guó)內(nèi)的阿里就是 Flink 的重度使用和貢獻(xiàn)者。Flink 是原生的流處理系統(tǒng)，把所有的數(shù)據(jù)都看成是流，認(rèn)為批處理是流處理中的一種特殊情況。數(shù)據(jù)基于 Flink Stream Source 流入，中間經(jīng)過(guò) Operator，從 Sink 流出。

為了解決流處理的容錯(cuò)問(wèn)題，F(xiàn)link 巧妙地運(yùn)用了分布式快照的設(shè)計(jì)與可部分重發(fā)的數(shù)據(jù)源實(shí)現(xiàn)容錯(cuò)。用戶可自定義對(duì)整個(gè) Job 進(jìn)行快照的時(shí)間間隔。當(dāng)任務(wù)失敗時(shí)，F(xiàn)link 會(huì)將整個(gè) Job 恢復(fù)到最近一次快照，并從數(shù)據(jù)源重發(fā)快照之后的數(shù)據(jù)。Flink 同時(shí)保證了實(shí)時(shí)性和吞吐量，流式語(yǔ)義也做得非常好，能夠保證 Exactly Once。

在此之外，組件技術(shù)選型的時(shí)候在滿足自己業(yè)務(wù)現(xiàn)狀的同時(shí), 還需要從以前幾個(gè)方面考慮:

• 開(kāi)源組件是否能覆蓋需求
• 開(kāi)源組件的擴(kuò)展性和二次開(kāi)發(fā)的難度
• 開(kāi)源組件 API 是否穩(wěn)定
• 開(kāi)源組件是否有應(yīng)用于生產(chǎn)環(huán)境的案例，比如多少公司應(yīng)用于生產(chǎn)環(huán)境
• 開(kāi)源組件社區(qū)是否活躍，比如可以看 github，issues，jiras 這些活躍程度
• 開(kāi)源組件 License 限定問(wèn)題
• 開(kāi)源組件之間的耦合問(wèn)題

2. 設(shè)計(jì)

下圖描述了 MES 實(shí)時(shí)計(jì)算引擎處理數(shù)據(jù)的過(guò)程：

圖 3：MES Streaming

數(shù)據(jù)從 Kafka 源源不斷地過(guò)來(lái)形成數(shù)據(jù)流，用戶通過(guò) UI 配置的一些規(guī)則形成實(shí)時(shí)配置流，數(shù)據(jù)流和配置流進(jìn)入到實(shí)時(shí)計(jì)算引擎 Spark Streaming 后進(jìn)行聚合計(jì)算。計(jì)算出的實(shí)時(shí)數(shù)據(jù)寫入到 Redis，歷史數(shù)據(jù)入庫(kù)到 HBase。UI 目前通過(guò) Restful API 來(lái)獲取實(shí)時(shí)和歷史數(shù)據(jù)。

3. 演進(jìn)

關(guān)于 MES 實(shí)時(shí)計(jì)算的引擎，我們主要經(jīng)歷了兩次演進(jìn)。

***代 ：Spark Streaming + Redis + HBase

在設(shè)計(jì)***代 MES 時(shí)，我們希望可以支持秒級(jí)的計(jì)算，并且精確計(jì)算每一個(gè)用戶。所以在當(dāng)時(shí)的現(xiàn)狀下，我們綜合考慮選擇了 Spark Streaming。

這個(gè)方案計(jì)算出來(lái)的 UV 是比較精確的。但它有自己的局限性：

首先，這一套架構(gòu)用到的幾個(gè)組件其實(shí)對(duì)資源都比較依賴, 而且 SparkStreaming 對(duì)那種時(shí)不時(shí)的流量高峰的數(shù)據(jù)處理不是非常友好。數(shù)據(jù)先在 Spark Streaming 算好然后再入 Redis，***再入庫(kù)到 Hbase，數(shù)據(jù)鏈路比較長(zhǎng)，不好維護(hù)。

另外，***代 MES 只支持自助配置規(guī)則，有了規(guī)則才會(huì)實(shí)時(shí)計(jì)算。所以對(duì)于比較自由的 OLAP 交叉分析不友好。而且如果由于集群不穩(wěn)定等原因?qū)е碌娜蝿?wù)失敗數(shù)據(jù)少算, 那么不管是用戶自助提交 Presto 還是利用 Spark 批處理全量重算，都是一個(gè)消耗集群資源的過(guò)程。由于批處理重算需要一定的時(shí)間來(lái)完成對(duì)歷史數(shù)據(jù)的修復(fù)，這對(duì)一些需要數(shù)據(jù)準(zhǔn)確并及時(shí)提供的用戶不是非常友好。

我們考慮，在數(shù)據(jù)量大的情況下，我們是不是可以適當(dāng)犧牲 UV 精準(zhǔn)度的計(jì)算，來(lái)保障整個(gè)系統(tǒng)的性能和穩(wěn)定性。所以就引入了 Flink + Druid。

第二代：引入 Flink + Druid

剛才我們已經(jīng)簡(jiǎn)單了解過(guò) Flink，再來(lái)說(shuō)下 Druid。

Druid 是一個(gè)大數(shù)據(jù)實(shí)時(shí)查詢和分析的高容錯(cuò)、高性能的開(kāi)源分布式系統(tǒng)，用來(lái)快速處理大規(guī)模的數(shù)據(jù)，它能夠?qū)崿F(xiàn)對(duì)大量數(shù)據(jù)的快速查詢和分析，不足是存在一個(gè) 2% 的誤差。但事實(shí)上，在數(shù)據(jù)量非常大的情況下，2% 的誤差是可以接受的。后面我們又通過(guò) Yahoo 提供的 Data Sketch，實(shí)現(xiàn) UV 計(jì)算的精確調(diào)控，可以實(shí)現(xiàn)在 800w 之下的數(shù)據(jù)量，UV 都是準(zhǔn)確的。最終的計(jì)算結(jié)果通過(guò) restful 接口提供給 MES 獲取數(shù)據(jù)并展現(xiàn)。

圖 4：關(guān)于 Druid

Flink + Druid 部分主要是用來(lái)處理數(shù)據(jù)量大、維度多，且不需要精確到秒級(jí)的業(yè)務(wù)數(shù)據(jù)，比如 Page logdata、mobile page、以及 Server Push。在最近 15 天的數(shù)據(jù)是可以精確到分鐘級(jí)別查詢的，對(duì)于歷史數(shù)據(jù)，粒度越精確，持久化到 Druid 里面的數(shù)據(jù)量就越大。

在離線批量導(dǎo)入部分，目前 Druid 支持小時(shí)級(jí)以及 T+1 天級(jí)的數(shù)據(jù)校正。因?yàn)槿绻?Flink +Tranquility 實(shí)時(shí)攝取數(shù)據(jù)這一階段 task 有異常的話，會(huì)導(dǎo)致實(shí)時(shí)數(shù)據(jù)到 Druid 有丟失的情況出現(xiàn)。因此根據(jù) Lambda 架構(gòu)的思想，我們可以用小時(shí)級(jí)或者天級(jí)離線數(shù)據(jù)對(duì)丟失的數(shù)據(jù)進(jìn)行重算補(bǔ)全。

對(duì)比一下兩代計(jì)算引擎，F(xiàn)link + Druid 的引入很好地解決了因?yàn)榇罅繑?shù)據(jù)的 UV 計(jì)算帶來(lái)的壓力：

圖 5：兩代實(shí)時(shí)計(jì)算引擎

MES 優(yōu)化歷程

為了更好地滿足業(yè)務(wù)需求，提升整個(gè)系統(tǒng)的性能，我們也在不斷對(duì) MES 進(jìn)行優(yōu)化，主要包括實(shí)時(shí)計(jì)算集群、計(jì)算引擎、查詢接口和監(jiān)控方面。這里主要和大家分享兩點(diǎn)。

1. 實(shí)時(shí)計(jì)算集群優(yōu)化

• Spark，Druid，F(xiàn)link 集群框架版本升級(jí)及相關(guān)參數(shù)優(yōu)化;
• Redis，Hbase 節(jié)點(diǎn)擴(kuò)容和參數(shù)優(yōu)化;
• 集群網(wǎng)絡(luò)，Yarn，Mesos 等資源管理框架調(diào)整和優(yōu)化

2. 實(shí)時(shí)計(jì)算引擎優(yōu)化

數(shù)據(jù)結(jié)構(gòu)和計(jì)算邏輯

對(duì)于 Spark 來(lái)講，Prefer 原生數(shù)據(jù)類型以及數(shù)組結(jié)構(gòu)，對(duì)于指針類型以及嵌套的結(jié)構(gòu)處理起來(lái)性能不是非常友好。因此要注意這一點(diǎn)，妥善優(yōu)化自己的數(shù)據(jù)結(jié)構(gòu)。

計(jì)算邏輯的部分也要考慮好。比如寫 Redis 的時(shí)候是事先規(guī)劃好要存入 Redis 中的數(shù)據(jù)結(jié)構(gòu)來(lái)利用 Akka 并發(fā)每條來(lái)寫入，還是在 Streaming 中算好一批結(jié)果***來(lái)一次性寫入 Redis，這 2 種方式在性能上還是有很大區(qū)別的。

參數(shù)優(yōu)化

(1) 序列化方式首先是 Kyro 的方式，其次才是 Java，序列化的方式不同對(duì)網(wǎng)絡(luò)的傳輸以及處理起來(lái)的性能是有影響的。

(2)Spark 推測(cè)執(zhí)行機(jī)制。根據(jù)我們集群目前的現(xiàn)狀，有各種各樣的任務(wù)同時(shí)在跑，如果遇到集群資源使用高峰期，那么一個(gè) Spark 任務(wù)落在比較慢的節(jié)點(diǎn)上就會(huì)拖累整個(gè) Job 的執(zhí)行進(jìn)度。開(kāi)啟推測(cè)執(zhí)行之后，系統(tǒng)會(huì)把進(jìn)程慢的任務(wù)主動(dòng)殺死，然后重新產(chǎn)生一個(gè)相同的任務(wù)分配到資源充沛的節(jié)點(diǎn)上去快速完成它。

(3) 數(shù)據(jù)本地化。分布式計(jì)算有一個(gè)經(jīng)典的理念是：移動(dòng)數(shù)據(jù)不如移動(dòng)計(jì)算。比如說(shuō)我把一個(gè)任務(wù)分成很多并行的任務(wù)，有可能獲得的任務(wù)剛好需要處理的數(shù)據(jù)就在處理的節(jié)點(diǎn)上，也有可能不是。所以這里有一個(gè)本地化等待時(shí)間的參數(shù)可以控制數(shù)據(jù)本地化的處理等級(jí)并對(duì)性能產(chǎn)生很大影響。

另外還用一些關(guān)于并行度控制、JVM GC 方面的調(diào)優(yōu)就比較細(xì)節(jié)了，如果大家感興趣可以留言給我們交流。

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

馬蜂窩實(shí)時(shí)計(jì)算平臺(tái)的發(fā)展還需要不斷探索，未來(lái)我們主要會(huì)在以下幾個(gè)方面重點(diǎn)推進(jìn)：

1. 實(shí)時(shí)計(jì)算任務(wù)統(tǒng)一資源管理和任務(wù)調(diào)度

2. 支持復(fù)雜的實(shí)時(shí) SQL OLAP 計(jì)算

3. 實(shí)時(shí)數(shù)據(jù)血緣關(guān)系及監(jiān)控預(yù)警

4. 復(fù)雜實(shí)時(shí) CEP 規(guī)則系統(tǒng)

本文作者：董良，馬蜂窩大數(shù)據(jù)平臺(tái)研發(fā)技術(shù)專家。2017 年加入馬蜂窩，現(xiàn)負(fù)責(zé)馬蜂窩實(shí)時(shí)計(jì)算平臺(tái)和數(shù)據(jù)中臺(tái)服務(wù)。2008 年畢業(yè)于西安郵電大學(xué)，曾在 Talend、神州專車等公司工作，先后從事數(shù)據(jù)集成中間件，數(shù)據(jù)倉(cāng)庫(kù)，實(shí)時(shí)計(jì)算平臺(tái)等方向的研發(fā)工作。

感謝關(guān)注，歡迎大家掃描下方二維碼訂閱「馬蜂窩技術(shù)」內(nèi)容并推薦給更多熱愛(ài)技術(shù)的朋友，希望有更多機(jī)會(huì)和大家交流。

【本文是51CTO專欄作者馬蜂窩技術(shù)的原創(chuàng)文章，作者微信公眾號(hào)馬蜂窩技術(shù)(ID：mfwtech)】

戳這里，看該作者更多好文