今天主要給大家分享餓了么在實時計算平臺方面的一些演進經驗，整個實時平臺經歷了從無到有、快速發(fā)展、平臺化的階段，每個階段都面臨著不同的問題。

餓了么 BDI-大數據平臺研發(fā)團隊目前共有 20 人左右，主要負責離線&實時 Infra 和平臺工具開發(fā)。

其中包括 20+ 組件的開發(fā)和維護、2K+ Servers 運維及數據平臺周邊衍生工具研發(fā)&維護。離線 Infra 和平臺工具這一塊對外分享的比較多。

餓了么實時計算平臺整體規(guī)模和架構

首先介紹下目前餓了么實時計算平臺的整體規(guī)模：

• 4 Kafka 集群，單 Kafka 高峰100wmsg/s。
• 2 ELK 集群用來做日志檢索，全網索引量 86w/s。
• 4 Storm 集群，根據業(yè)務 SLA 進行物理拆分，全網高峰計算量 1.6kw/s。
• 2 Spark Streaming 集群和 2 個 Flink 集群，都是 On Yarn 模式，其中 Flink 在做一些線上業(yè)務嘗試。
• 20 個以上業(yè)務方接入，共 120+ Storm 任務（包括雙活 & DataPipeline 業(yè)務）、26+ Spark Streaming 任務、10+ Streaming SQL 任務，涉及實時搜索推薦、實時風控、實時監(jiān)控、實時營銷等項目。

餓了么實時計算平臺整體架構圖如下：

從上圖可以看到，包括了數據采集、傳輸、計算、落地還有服務。組件多且系統(tǒng)壓力大，部分業(yè)務直接應用于關鍵路徑，要求整體的 SLA 要在 99.99% 以上。

餓了么實時計算平臺演進過程

餓了么實時計算平臺的演講過程經歷了幾個階段：從無到有、快速發(fā)展、平臺化。

從無到有

餓了么從 2015 年 5 月份上線實時計算，這個階段面臨如下幾個問題：

• 需求少。實時業(yè)務在公司推廣不夠，只有一個 UBT Domain QoS 分析需求，用來計算網站錯誤率、HTTPCode 分布、頁面加載速度等 QoS 數據。
• 數據源單一。數據源只有用戶行為 Log，缺乏訂單、運單等核心 DB 數據。
• 容量有限。集群規(guī)模不到 20 臺，且和離線混部，要同時支持實時+離線需求，存在資源競爭。
• 穩(wěn)定性問題。缺乏統(tǒng)一的使用標準姿勢，各組件應用問題也比較多。比如每種應用需要什么類型的機器、怎么樣最佳化配置……同時因為使用的開源組件版本較舊，存在一些穩(wěn)定性 Bug。
• 數據延遲和數據不全。因為在初始技術選項上存在問題，在數據采集端使用了自行開發(fā)的 Python 程序。

同時使用了跨機房的傳輸，導致經常出現(xiàn)數據丟失和延遲問題。最嚴重時，實時的數據在高峰期要 Delay 2 個小時以上，基本不可用。

這個階段主要解決了環(huán)境、穩(wěn)定性、數據延遲和數據丟失的問題，主要做了如下工作。

針對環(huán)境和標準化問題：

• 根據業(yè)務特點確定新機型配置（區(qū)分計算密集型、內存密集型、IO 密集型、混合密集型等）。
• 設定各組件部署規(guī)范&參數規(guī)范，包括環(huán)境變量、內核參數、應用參數等：比如對 JVM 參數的調整、對 ZK/Kafka 的配置優(yōu)化。
• 包括硬件的標準：比如舊的 Kafka 不支持 JBOD Markdown ，因此我們會對磁盤做 RAID，增強系統(tǒng)的可用性。
• 實時和離線進行拆分，獨立部署，避免資源的爭用。
• 對現(xiàn)有組件升級，修復穩(wěn)定性 Bug（比如 Storm 版本演進 0.9.4->0.9.6 到后面的 1.0.1、1.0.3）。
• 同時基于環(huán)境的標準化和自動化考慮，引入了 Puppet 做環(huán)境配置收斂保持，Ansible 做自動化部署，并做成一鍵管理工具，可以快速地維護。

針對穩(wěn)定性問題：

• 在數據采集側，為了減少對 App 端的流量消耗，同時增強整體傳輸的效率，在 App 端引入 Merge+Gzip，在 SDK 把多條消息 Merge 成一條壓縮后發(fā)送到 Nginx Server，并在 Nginx 端用 Lua 進行解壓，以 AccessLog 的方式落地。
• 在數據傳輸側，重新調研數據傳輸方案，考慮到團隊以 Java 為技術棧，以及外部案例，引入了 Flume 作為數據采集管道。

以 Tail Log 的形式進行數據采集并 Sink 到 Kafka 集群，并基于 HDFS Sink 開發(fā)根據 EventTime 的 Partition 功能，同時 fix Backlog 和 Kafka Sink 的 Bug。

• 在數據落地側，為了存儲中間狀態(tài)結果，引入 KV 存儲。最初使用單機 Redis 存儲數據，set 去重，遇到了嚴重的性能問題，所以后面開始逐步使用 Self Sharding->Redis+Tewmproxy 的方式，但是維護成本比較高。

后面隨著 Redis Cluster 穩(wěn)定版 Release 開始逐步遷移到 Cluster 模式，這個階段公司在 NoSQL 的經驗一直不足，所以都是團隊內部不斷地摸索前進。

這個階段實時架構如下圖所示：

這個階段整個平臺研發(fā)只有 4 個人，既要負責離線、實時、平臺工具的開發(fā)和維護，還要支撐業(yè)務的開發(fā)，資源比較緊張，實時方面投入捉襟見肘。

雖然解決了基本的穩(wěn)定性 & 數據延遲和丟失的問題，但是整體鏈路 SLA 還是不高，同時還存在數據源單一、應用單一的問題。

快速發(fā)展

2016 年公司業(yè)務大力發(fā)展，實時方面的需求越來越多。SLA 不高、數據源單一、應用單一的問題亟待解決。

由于業(yè)務需求，需開發(fā)實時 Dashboard 來實時關注業(yè)務情況，涉及流量、訂單、運單等重要數據，項目需要涉及不同的數據源（Log & DB &業(yè)務數據），同時要求 SLA 99.99% 以上。

為了提高整體的 SLA，同時覆蓋 DB 側的數據源，針對整個鏈路做了如下調整優(yōu)化：

數據源方面：

• 優(yōu)化 UBT（Nginx AccessLog）傳輸效率，調整 Flume 和 Kafka 的 Batch 大小、增加 Flume Timeout send 功能（默認 Flume1.6.0只有 Batch 的控制，沒有 Timeout 的控制，會導致低峰時消息 Delay 變大）?？紤]到 Batch Size 越大吞吐量越高，相應延遲也越大，因此需要針對吞吐量 & 實時性要求做 Trade Off。

通過線上數據的分析，最終把 Batch Size 調整到 10，Timeout 調整到100ms，最終 UBT 數據鏈路（從采集 Server 到計算落地）延遲降低 99.9%<1s。

• 同時為了防止異常流量導致日志收集端雪崩，引入了 Nginx 限流限速模塊。
• 為了滿足 Binlog 數據采集的需求引入 OR 做解析工具，為防止 OR 異常退出導致數據丟失問題，開發(fā)基于 ZK 存儲 Offset 功能，在異常 Crash 重啟后繼續(xù)上次的 Offset 消費即可。

計算方面：

• 前面提到用戶 Log 是合并發(fā)送，在 Kafka 中的表現(xiàn)是多條 Merge 成一條，應用如果需要使用的話，需要按照一定的規(guī)則 Split。

同時每個業(yè)務關注的 Type 不一樣，不同的業(yè)務需要全量消費所有 Log，同時需要自行進行 Split，計算量大，維護成本比較高。

為了解決這個問題引入了雙層 Kafka 結構，在第一層進行統(tǒng)一的 Split 和 Filter，過濾異常流量，同時分 Type 寫入二層 Topic，這樣每個消費方只需消費對應的數據部分即可，整體流量相關業(yè)務計算量對比之前降低了一半以上。

• 涉及 UV 計算的場景，初始使用 Redis Set 去重，但是內存消耗過大。由于 UV 指標允許 1% 以內誤差，在精度和時空效率上做 Trade Off，轉而使用 Redis 的 HLL 來估算。

隨著業(yè)務量的增大，Redis 的 QPS 成為瓶頸，同時 Redis 無法跨實例進行 HLL 的 Merge，又演化為基于內存的 HLL 估算和 Merge，同時使用 Redis 直接存儲對象，節(jié)省百倍內存的同時，支持多維度的 Merge 操作。

• 同時考慮到多個系統(tǒng)共用 ZK，ZK 可能存在比較大的壓力，因此通過分析 ZK 的 Transaction Log 來確定調用分布。

比如通過分析發(fā)現(xiàn) Storm Worker 的 Heartbeat 會頻繁訪問 ZK，因此通過增加 Heartbeat Commit 時間減少 ZK 的壓力。

• 為了減少重復的代碼開發(fā)，對基礎組件進行了封裝：包括數據消費、去重、累加、數據寫入等算子，最終減少了部分任務 50% 的代碼量，提高了整體的開發(fā)效率，讓用戶關注業(yè)務邏輯即可。

封裝組件列表

運維管理方面：

• 為了解整體的容量情況，開發(fā)實時容量看板，通過實時獲取 Zabbix Item LastValue 來監(jiān)控 Storm & Redis Cluster 實時壓力情況。
• 為方便用戶可以快速查看任務 Log 引入 ELK，同時在 Kafka2es 這一層引入 Hangout 替代 Flume （可支持 3x 以上性能提升）。

最終實現(xiàn)了 Storm Top Log->Logstash->Kafka->Hangout->ES->Kanbana 的整個 Log 鏈路。

整體 SLA 增強方面：

• 數據采集側和計算側均引入限速 & 反壓功能，防止流量暴漲導致的雪崩效應。
• 從數據采集到數據最終落地使用雙鏈路的方式，并采用多機房方式：比如數據采集端分布在異地兩機房，使用本地計算+最終 Merge 或者數據雙向傳輸+全量計算的方式。
• 配合 Kafka Binlog 數據的重放和 Nginx 模擬訪問，對全鏈路進行模擬壓測，配合實時監(jiān)控 & 業(yè)務表現(xiàn)來關注各組件的性能臨界點。
• 基于 3 的數據進行容量規(guī)劃，同時做好資源的隔離。
• 完善各層面的監(jiān)控和報警策略，包括模擬用戶的訪問來驗證鏈路的情況，同時為了避免口徑不一致導致的數據問題，開發(fā)離線&實時數據質量對比 Report，監(jiān)控數據質量問題。
• 在業(yè)務端&后端增加降級策略，同時后端服務 Auto Failover。
• 完善各緊急預案 SOP，并針對性演練：比如通過切斷 Active 的數據落地層來確定應用是否可以 Auto Failover。
• 數據層利用緩存，比如把一些 Storm 中間計算態(tài)數據寫入 Redis 中。
• 同時為了防止應用 Crash 導致數據斷層問題，開發(fā)了補數功能，在異?；謴秃罂梢酝ㄟ^離線的數據補齊歷史數據。

通過上述的一系列調整，最終抗住業(yè)務幾倍的流量增長，保證了整體服務的穩(wěn)定性。

業(yè)務監(jiān)控 Dashboard 部分示例，如下圖：

這個階段實時平臺的主要用戶還是大數據自身，應用架構如下：

這個階段雖然解決了數據源單一、整體 SLA 不高的問題，但是也帶來了新的問題：

• Storm 任務越來越多、Kafka Topic 越來越多、Topic 的 Producer 端和 Consumer 各是哪些、數據量如何，同時任務維護成本也越來越高，亟需一個平臺化的工具。
• 用戶的需求多樣，一種引擎已經無法滿足業(yè)務需求，需要多種引擎支持。
• 由于餓了么存在多語言開發(fā)場景（Go、Python、Java、Scala 等），Storm 任務開發(fā)學習成本比較高，用戶希望通過一個 SQL 即可支持部分簡單場景。
• 業(yè)務 Log 需要通過 Nginx+Flume 方式收集，需要協(xié)調 OPS 部署 Flume，而 OPS 的同學比較排斥 Flume，亟需提供一個方便開發(fā)數據寫入的標準 SDK 接入姿勢。
• 之前業(yè)務數據落地倉庫需要通過 Kafka->Flume->HDFS 方式，隨著接入業(yè)務方變多，F(xiàn)lume 的維護成本越來越高，同時發(fā)生了多起 2HDFS 的 Bug 導致數據重復的問題，亟需提供一個方便維護、支持多種數據 Sink 方式、穩(wěn)定的數據流通解決方案。

平臺化

2017 年初各產研逐步接入實時計算，上述問題也逐漸暴露出來，平臺層面亟需一個統(tǒng)一的方案來解決用戶的痛點。

因此在年初，我們確定了“以 ERDP 實時平臺為核心，打通數據采集、數據傳輸、數據計算、數據落地 DataPipeline 整體流程，為用戶提供一個一站式的實時平臺”的方向。

在此目標之上，我們做了如下的調整：

開發(fā)資源聚焦：

• 因為負責實時平臺相關的小伙伴只有 3-5 人，我們開始逐步推動部分組件接入公司統(tǒng)一服務（比如監(jiān)控 & 報警），并將部分任務移交到應用團隊 team，只保留個別實時業(yè)務和平臺相關部分開發(fā)。
• 考慮到 OR 每接入一個 DB 都需要啟動 Instance，維護成本比較高，同時公司開始推動統(tǒng)一的 Binlog 解析工具 DRC 來做多機房間 DB 數據同步，因此我們也逐步使用 DRC 來替換 OR。

解決數據采集痛點：

• 通過開發(fā)統(tǒng)一的多語言數據接入 SDK 并接入配置中心，標準化數據接入的格式，方便用戶的業(yè)務數據可以快速接入，同時為了提高可用性，在 SDK 中集成了降級 & 限速功能。

解決數據傳輸接入痛點：

• 參考 Flume Source->Channel->Sink 的思想，開發(fā)了基于 Storm 的 Data Pipeline 功能的組件 EDSink，可以支持多種數據寫入方式，包括 2KAFKA、2ES、2HDFS、2DB。

同時高度封裝，抽象出部分配置，用戶只需要通過填寫一部分參數就可以實現(xiàn)數據的落地，大大加快了數據接入的效率，同時在數據落地層面引入了熔斷功能。

• 在數據通過 EDSink 寫入 HDFS 方面，打通了離線平臺的調度系統(tǒng)和元數據管理使用，集成了建表和數據清洗功能，實現(xiàn)了一鍵化的數據落地。

目前業(yè)務 Log 通過 EDSink 接入數據倉庫已經從之前的 2-3 天降低到 2 小時以內。

• 為支持更細粒度的任務調度，在 EDSink 中集成基于 EventTime 分區(qū)功能，可以支持分鐘粒度分區(qū)，結合 Spark 來支持半小時 ETL 鏈路的開發(fā)，小時整體鏈路從之前的 40min 縮短到 20min 左右即可完成。
• 同時和 Binlog 解析工具聯(lián)動打通，支持用戶自助申請落地 DB 數據，目前基于此方案，團隊在進行 DB 數據去 Sqoop 化，預計可大大節(jié)省線上 DB Slave 服務器成本。

提供更多的計算方式：

• 引入 Spark Streaming 并集成到 ERDP 平臺，封裝基本的 Spark Streaming 算子，用戶可以通過平臺對 Spark Streaming 任務進行管理。
• 考慮到需要支持部分 SQL 的需求，在 Spark Streaming、Flink、Storm CQL 等引擎中做了對比，從團隊的技術棧、引擎的成熟度、穩(wěn)定性等層面綜合考慮最終選擇了 Spark Streaming。

并基于 Spark Streaming 的 SQL 功能，為用戶封裝基本算子，同時支持上傳 Jar 包提供 UDF 功能及 Scala 腳本支持，支持 Structured Streaming 以支持帶狀態(tài)的增量計算，實現(xiàn)用戶寫 SQL 即可滿足實時開發(fā)的需求（目前可支持 90% 的業(yè)務場景）。

自動化&自助化便于任務和資源管理：

• 之前用戶的 Storm 任務配置更改都需要重新打包，用戶管理成本比較高，因此引入 Storm Flux 功能，封裝基本組件并集成到實時平臺。

用戶可以 0 代碼生成數據 Sink，同時可以快速進行任務的開發(fā)和管理，自動生成 YML 文件降低任務的維護成本。

• 通過打通各個資源申請流程，支持 Kafka Topic 等資源的自助化申請和自動化創(chuàng)建，基于 Topic 數據完善元數據的管理，為資源的核算和實時元數據血緣做數據基礎。
• 為了方便任務的監(jiān)控，將 Storm，Spark Streaming，Kafka 層面監(jiān)控統(tǒng)一入庫 InfluxDB，并自動化模板生成功能，用戶無需手動添加監(jiān)控和報警。

任務上線后 Metric & Dashboard 自動上報和創(chuàng)建，通過自動采集 API 的數據寫入 InfluxDB，同時做了一個標準的 Template 用來自動生成 Grafana 的監(jiān)控模板。

Kafka 監(jiān)控示例

通過上述一系列的調整，最終完善了整個平臺，解決了用戶開發(fā)成本高、接入成本高、管理成本高等痛點，最終的架構圖就是文章開始的狀況。

后續(xù)計劃

雖然經過了一些演進，現(xiàn)有的平臺仍然存在一些問題，比如：

• SQL 方式覆蓋場景有限。
• 用戶在引擎中選擇困難，沒有一個引擎可以解決大部分需求。
• Kafka0.8.2 版本功能有限，不支持 Excatly Once，不支持 JBOD Markdown 等。
• 實時和離線分離，數據重復建設，因為實現(xiàn)方式不同，實時和離線很難做到數據口徑完全一致。
• 實時業(yè)務場景在公司內覆蓋不夠。
• ……

因此針對這些痛點，我們也在做如下嘗試：

• Flink 以其性能&易用性在實時計算領域開始大行其道，我們也在做一些業(yè)務試點。
• 實時和離線融合 CEP 場景的試水。
• Kafka 新版本的引入，包括 Qouta 限速、JBOD Markdown、Stream API、Excatly Once 等功能的支持。
• 實時平臺集成到統(tǒng)一的一體化平臺，用戶在一個平臺完成實時 & 離線開發(fā)。
• 發(fā)掘線上的業(yè)務場景，比如我們目前和策略部門合作的實時營銷項目就是通過用戶的行為數據來做一些策略，提高轉化率。

平臺化演進中的一些心得

最后說一下關于平臺化演進中的心得：

• 學會資源聚集（善于借助外力，must to have VS nice to have）。
• MVP 最小化可行性調研（產品是迭代出來的，不是一蹴而就的，先完成后完美）。
• 偷懶思想，不重復造輪子（他山之石可以攻玉）。
• 賦能用戶，盡量自動化&自助化（提高效率，解放生產力）。
• 數據化運營（用數據說話）。
• 資源隔離，重點業(yè)務的 SLA 保證（降級限流策略，反壓功能等）。
• 做好監(jiān)控（全鏈路的監(jiān)控，并做必要測試）。
• 防范墨菲定律，及時做好容量規(guī)劃 & 壓測，同時不斷完善 SOP。
• 抽象思維（從一個到一類問題的抽象）。
• 以解決實際問題為目的，而不是炫技。
• 關注用戶體驗（做完了 VS 做好了）。
• 防火勝于救火，多想幾步，不斷總結并完善標準 & 規(guī)劃 & 流程（上線規(guī)范/ Checklist / SOP 流程等）。