餓了么BDI-大數(shù)據(jù)平臺研發(fā)團隊目前共有20人左右，主要負責離線&實時 Infra 和平臺工具開發(fā)，其中包括20+組件的開發(fā)和維護、2K+ Servers 運維及數(shù)據(jù)平臺周邊衍生工具研發(fā)&維護。離線 Infra 和平臺工具這一塊對外分享的比較多。

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

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

• 4 Kafka 集群，單 Kafka 高峰100wmsg/s；

• 2 ELK 集群用來做日志檢索，全網(wǎng)索引量86w/s；

• 4 Storm 集群，根據(jù)業(yè)務(wù) SLA 進行物理拆分，全網(wǎng)高峰計算量1.6kw/s；

• 2 Spark Streaming 集群和2個 Flink 集群，都是 On Yarn 模式，其中 Flink 在做一些線上業(yè)務(wù)嘗試；

• 20個以上業(yè)務(wù)方接入，共120+ Storm 任務(wù)（包括雙活 & DataPipeline 業(yè)務(wù)）、26+ Spark Streaming 任務(wù)、10+ Streaming SQL 任務(wù)，涉及實時搜索推薦、實時風(fēng)控、實時監(jiān)控、實時營銷等項目。

整體架構(gòu)圖如下：

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

整體也經(jīng)歷幾個階段：從無到有、快速發(fā)展、平臺化。

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

• 需求少。實時業(yè)務(wù)在公司推廣不夠，只有一個 UBT Domain QoS 分析需求，用來計算網(wǎng)站錯誤率、HTTPCode 分布、頁面加載速度等 QoS 數(shù)據(jù)；

• 數(shù)據(jù)源單一。數(shù)據(jù)源只有用戶行為 Log，缺乏訂單、運單等核心 DB 數(shù)據(jù)；

• 容量有限。集群規(guī)模不到20臺，且和離線混部，要同時支持實時+離線需求，存在資源競爭；

• 穩(wěn)定性問題。缺乏統(tǒng)一的使用標準姿勢，各組件應(yīng)用問題也比較多。比如每種應(yīng)用需要什么類型的機器、怎么樣最佳化配置……同時因為使用的開源組件版本較舊，存在一些穩(wěn)定性 bug；

• 數(shù)據(jù)延遲和數(shù)據(jù)不全。因為在初始技術(shù)選項上存在問題，在數(shù)據(jù)采集端使用了自行開發(fā)的 Python 程序，同時使用了跨機房的傳輸，導(dǎo)致經(jīng)常出現(xiàn)數(shù)據(jù)丟失和延遲問題。最嚴重時，實時的數(shù)據(jù)在高峰期要 Delay 2個小時以上，基本不可用。

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

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

• 根據(jù)業(yè)務(wù)特點確定新機型配置（區(qū)分計算密集型、內(nèi)存密集型、IO 密集型、混合密集型等）；

• 設(shè)定各組件部署規(guī)范&參數(shù)規(guī)范，包括環(huán)境變量、內(nèi)核參數(shù)、應(yīng)用參數(shù)等：比如對 JVM 參數(shù)的調(diào)整、對 ZK/Kafka 的配置優(yōu)化；包括硬件的標準：比如舊的 Kafka 不支持 JBOD Markdown ，因此我們會對磁盤做 RAID，增強系統(tǒng)的可用性；

• 實時和離線進行拆分，獨立部署，避免資源的爭用；

• 對現(xiàn)有組件升級，修復(fù)穩(wěn)定性 bug（比如 Storm 版本演進0.9.4->0.9.6到后面的1.0.1、1.0.3）；

  
  

• 同時基于環(huán)境的標準化和自動化考慮，引入了 Puppet 做環(huán)境配置收斂保持，Ansible 做自動化部署，并做成一鍵管理工具，可以快速地維護。

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

• 在數(shù)據(jù)采集側(cè)，為了減少對 app端的流量消耗，同時增強整體傳輸?shù)男?，?app端引入 Merge+Gzip，在 SDK 把多條消息 Merge 成一條壓縮后發(fā)送到 Nginx Server，并在 Nginx 端用 Lua 進行解壓，以 AccessLog 的方式落地；

• 在數(shù)據(jù)傳輸側(cè)，重新調(diào)研數(shù)據(jù)傳輸方案，考慮到團隊以 Java 為技術(shù)棧，以及外部案例，引入了 Flume 作為數(shù)據(jù)采集管道，以 Tail Log 的形式進行數(shù)據(jù)采集并 Sink 到 Kafka 集群，并基于 HDFS Sink 開發(fā)根據(jù) EventTime 的 Partition 功能，同時 fix Backlog 和 Kafka Sink 的 bug；

  
  

• 在數(shù)據(jù)落地側(cè)，為了存儲中間狀態(tài)結(jié)果，引入 KV 存儲。最初使用單機 Redis 存儲數(shù)據(jù)，set 去重，遇到了嚴重的性能問題，所以后面開始逐步使用 Self Sharding->Redis+Tewmproxy 的方式，但是維護成本比較高。后面隨著 RedisCluster 穩(wěn)定版 Release 開始逐步遷移到 Cluster 模式，這個階段公司在 NoSQL 的經(jīng)驗一直不足，所以都是團隊內(nèi)部不斷地摸索前進。

這個階段實時架構(gòu)是這樣的：

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

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

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

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

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

數(shù)據(jù)源方面：

• 優(yōu)化 UBT（Nginx AccessLog）傳輸效率，調(diào)整 Flume 和 Kafka 的 Batch 大小、增加 Flume Timeout send 功能（默認 Flume1.6.0只有 Batch 的控制，沒有 Timeout 的控制，會導(dǎo)致低峰時消息 Delay 變大）?？?/span>慮到 Batch Size 越大吞吐量越高，相應(yīng)延遲也越大，因此需要針對吞吐量 & 實時性要求做 Trade Off。通過線上數(shù)據(jù)的分析，最終把 Batch Size 調(diào)整到10，Timeout 調(diào)整到100ms，最終 UBT 數(shù)據(jù)鏈路（從采集 Server 到計算落地）延遲降低99.9%<>

• 同時為了防止異常流量導(dǎo)致日志收集端雪崩，引入了 Nginx 限流限速模塊；

• 為了滿足 Binlog 數(shù)據(jù)采集的需求引入 OR 做解析工具，為防止 OR 異常退出導(dǎo)致數(shù)據(jù)丟失問題，開發(fā)基于 ZK 存儲 Offset 功能，在異常 Crash 重啟后繼續(xù)上次的 Offset 消費即可。

計算方面：

• 前面提到用戶 Log 是合并發(fā)送，在 Kafka 中的表現(xiàn)是多條 Merge 成一條，應(yīng)用如果需要使用的話，需要按照一定的規(guī)則 Split。同時每個業(yè)務(wù)關(guān)注的 Type 不一樣，不同的業(yè)務(wù)需要全量消費所有 Log，同時需要自行進行 Split，計算量大，維護成本比較高。為了解決這個問題引入了雙層 Kafka 結(jié)構(gòu)，在第一層進行統(tǒng)一的 Split 和 Filter，過濾異常流量，同時分 Type 寫入二層 Topic，這樣每個消費方只需消費對應(yīng)的數(shù)據(jù)部分即可，整體流量相關(guān)業(yè)務(wù)計算量對比之前降低了一半以上；

  
  

• 涉及 UV 計算的場景，初始使用 Redis Set 去重，但是內(nèi)存消耗過大。由于 UV 指標允許1%以內(nèi)誤差，在精度和時空效率上做 Trade Off，轉(zhuǎn)而使用 Redis 的 HLL 來估算。隨著業(yè)務(wù)量的增大，Redis 的 QPS 成為瓶頸，同時 Redis 無法跨實例進行 HLL 的 Merge，又演化為基于內(nèi)存的 HLL 估算和 Merge，同時使用 Redis 直接存儲對象，節(jié)省百倍內(nèi)存的同時，支持多維度的 Merge 操作；

• 同時考慮到多個系統(tǒng)共用 ZK，ZK 可能存在比較大的壓力，因此通過分析 ZK 的 Transcation Log 來確定調(diào)用分布。比如通過分析發(fā)現(xiàn) Storm Worker 的 Heartbeat 會頻繁訪問 ZK，因此通過增加 Heartbeat Commit 時間減少 ZK 的壓力；

  
  

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

封裝組件列表

運維管理方面：

• 為了解整體的容量情況，開發(fā)實時容量看板，通過實時獲取 Zabbix Item LastValue 來監(jiān)控 Storm & RedisCluster 實時壓力情況；

  
  

• 為方便用戶可以快速查看任務(wù) Log 引入 ELK，同時在 Kafka2es 這一層引入 Hangout 替代 Flume （可支持3x以上性能提升），最終實現(xiàn)了 Storm Top Log->Logstash->Kafka->Hangout->ES->Kanbana 的整個 Log 鏈路。

整體 SLA 增強方面：

• 數(shù)據(jù)采集側(cè)和計算側(cè)均引入限速 & 反壓功能，防止流量暴漲導(dǎo)致的雪崩效應(yīng)；

  
  

• 從數(shù)據(jù)采集到數(shù)據(jù)最終落地使用雙鏈路的方式，并采用多機房方式：比如數(shù)據(jù)采集端分布在異地兩機房，使用本地計算+最終 Merge 或者數(shù)據(jù)雙向傳輸+全量計算的方式；

• 配合 Kafka Binlog 數(shù)據(jù)的重放和 Nginx 模擬訪問，對全鏈路進行模擬壓測，配合實時監(jiān)控 & 業(yè)務(wù)表現(xiàn)來關(guān)注各組件的性能臨界點；

• 基于3的數(shù)據(jù)進行容量規(guī)劃，同時做好資源的隔離；

• 完善各層面的監(jiān)控和報警策略，包括模擬用戶的訪問來驗證鏈路的情況，同時為了避免口徑不一致導(dǎo)致的數(shù)據(jù)問題，開發(fā)離線&實時數(shù)據(jù)質(zhì)量對比 Report，監(jiān)控數(shù)據(jù)質(zhì)量問題；

• 在業(yè)務(wù)端&后端增加降級策略，同時后端服務(wù) Auto Failover；

• 完善各緊急預(yù)案 SOP，并針對性演練：比如通過切斷 Active 的數(shù)據(jù)落地層來確定應(yīng)用是否可以 Auto Failover；

• 數(shù)據(jù)層利用緩存，比如把一些 Storm 中間計算態(tài)數(shù)據(jù)寫入 Redis 中；

  
  

• 同時為了防止應(yīng)用 Crash 導(dǎo)致數(shù)據(jù)斷層問題，開發(fā)了補數(shù)功能，在異?；謴?fù)后可以通過離線的數(shù)據(jù)補齊歷史數(shù)據(jù)。

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

業(yè)務(wù)監(jiān)控 Dashboard 部分示例：

這個階段實時平臺的主要用戶還是大數(shù)據(jù)自身，應(yīng)用架構(gòu)如下：

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

• Storm 任務(wù)越來越多、Kafka Topic 越來越多、Topic 的 Producer 端和 Consumer 各是哪些、數(shù)據(jù)量如何，同時任務(wù)維護成本也越來越高，亟需一個平臺化的工具；

• 用戶的需求多樣，一種引擎已經(jīng)無法滿足業(yè)務(wù)需求，需要多種引擎支持；

• 由于餓了么存在多語言開發(fā)場景（Go、Python、Java、Scala 等），Storm 任務(wù)開發(fā)學(xué)習(xí)成本比較高，用戶希望通過一個 SQL 即可支持部分簡單場景；

• 業(yè)務(wù) Log 需要通過 Nginx+Flume 方式收集，需要協(xié)調(diào) OPS 部署 Flume，而 OPS 的同學(xué)比較排斥 Flume，亟需提供一個方便開發(fā)數(shù)據(jù)寫入的標準 SDK 接入姿勢；

• 之前業(yè)務(wù)數(shù)據(jù)落地倉庫需要通過 Kafka->Flume->HDFS 方式，隨著接入業(yè)務(wù)方變多，F(xiàn)lume 的維護成本越來越高，同時發(fā)生了多起 2HDFS 的 bug 導(dǎo)致數(shù)據(jù)重復(fù)的問題，亟需提供一個方便維護、支持多種數(shù)據(jù) Sink 方式、穩(wěn)定的數(shù)據(jù)流通解決方案。

17年初各產(chǎn)研逐步接入實時計算，上述問題也逐漸暴露出來，平臺層面亟需一個統(tǒng)一的方案來解決用戶的痛點。因此在年初，我們確定了“以 ERDP 實時平臺為核心，打通數(shù)據(jù)采集、數(shù)據(jù)傳輸、數(shù)據(jù)計算、數(shù)據(jù)落地 DataPipeline 整體流程，為用戶提供一個一站式的實時平臺”的方向。

在此目標之上，我們做了如下的調(diào)整：

開發(fā)資源聚焦：

• 因為負責實時平臺相關(guān)的小伙伴只有3-5人，我們開始逐步推動部分組件接入公司統(tǒng)一服務(wù)（比如監(jiān)控 & 報警），并將部分任務(wù)移交到應(yīng)用團隊 team，只保留個別實時業(yè)務(wù)和平臺相關(guān)部分開發(fā)；

• 考慮到 OR 每接入一個 DB 都需要啟動 Instance，維護成本比較高，同時公司開始推動統(tǒng)一的 Binlog 解析工具 DRC 來做多機房間 DB 數(shù)據(jù)同步，因此我們也逐步使用 DRC 來替換 OR。

解決數(shù)據(jù)采集痛點：

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

解決數(shù)據(jù)傳輸接入痛點：

• 參考 Flume Source->Channel->Sink 的思想，開發(fā)了基于 Storm 的 Data Pipeline 功能的組件 EDSink，可以支持多種數(shù)據(jù)寫入方式，包括 2KAFKA、2ES、2HDFS、2DB。同時高度封裝，抽象出部分配置，用戶只需要通過填寫一部分參數(shù)就可以實現(xiàn)數(shù)據(jù)的落地，大大加快了數(shù)據(jù)接入的效率，同時在數(shù)據(jù)落地層面引入了熔斷功能；

• 在數(shù)據(jù)通過 EDSink 寫入 HDFS 方面，打通了離線平臺的調(diào)度系統(tǒng)和元數(shù)據(jù)管理使用，集成了建表和數(shù)據(jù)清洗功能，實現(xiàn)了一鍵化的數(shù)據(jù)落地。目前業(yè)務(wù) Log 通過 EDSink 接入數(shù)據(jù)倉庫已經(jīng)從之前的2-3天降低到2小時以內(nèi)；

• 為支持更細粒度的任務(wù)調(diào)度，在 EDSink 中集成基于 EventTime 分區(qū)功能，可以支持分鐘粒度分區(qū)，結(jié)合 Spark 來支持半小時 ETL 鏈路的開發(fā)，小時整體鏈路從之前的40min縮短到20min左右即可完成；

  
  

• 同時和 Binlog 解析工具聯(lián)動打通，支持用戶自助申請落地 DB 數(shù)據(jù)，目前基于此方案，團隊在進行 DB 數(shù)據(jù)去 Sqoop 化，預(yù)計可大大節(jié)省線上 DB Slave 服務(wù)器成本。

提供更多的計算方式：

• 引入 Spark Streaming 并集成到 ERDP 平臺，封裝基本的 Spark Streaming 算子，用戶可以通過平臺對 Spark Streaming 任務(wù)進行管理；

  
  

• 考慮到需要支持部分 SQL 的需求，在 Spark Streaming、Flink、Storm CQL 等引擎中做了對比，從團隊的技術(shù)棧、引擎的成熟度、穩(wěn)定性等層面綜合考慮最終選擇了 Spark Streaming。并基于 Spark Streaming 的 SQL 功能，為用戶封裝基本算子，同時支持上傳 Jar 包提供 UDF 功能及 Scala 腳本支持，支持 Structured Streaming 以支持帶狀態(tài)的增量計算，實現(xiàn)用戶寫 SQL 即可滿足實時開發(fā)的需求（目前可支持90%的業(yè)務(wù)場景）。

自動化&自助化便于任務(wù)和資源管理：

• 之前用戶的 Storm 任務(wù)配置更改都需要重新打包，用戶管理成本比較高，因此引入 Storm Flux 功能，封裝基本組件并集成到實時平臺。用戶可以0代碼生成數(shù)據(jù) Sink，同時可以快速進行任務(wù)的開發(fā)和管理，自動生成 YML 文件降低任務(wù)的維護成本；

• 通過打通各個資源申請流程，支持 Kafka Topic 等資源的自助化申請和自動化創(chuàng)建，基于 Topic 數(shù)據(jù)完善元數(shù)據(jù)的管理，為資源的核算和實時元數(shù)據(jù)血緣做數(shù)據(jù)基礎(chǔ)；

  
  

• 為了方便任務(wù)的監(jiān)控，將 Storm，SparkStreaming，Kafka 層面監(jiān)控統(tǒng)一入庫 InfluxDB，并自動化模板生成功能，用戶無需手動添加監(jiān)控和報警，任務(wù)上線后 Metric & Dashboard 自動上報和創(chuàng)建，通過自動采集 API 的數(shù)據(jù)寫入 InfluxDB，同時做了一個標準的 Template 用來自動生成 Grafana 的監(jiān)控模板。

Kafka監(jiān)控示例

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

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

• SQL 方式覆蓋場景有限；

• 用戶在引擎中選擇困難，沒有一個引擎可以解決大部分需求；

• Kafka0.8.2 版本功能有限，不支持 Excatly Once，不支持 JBOD Markdown 等；

• 實時和離線分離，數(shù)據(jù)重復(fù)建設(shè)，因為實現(xiàn)方式不同，實時和離線很難做到數(shù)據(jù)口徑完全一致；

• 實時業(yè)務(wù)場景在公司內(nèi)覆蓋不夠；

• ……
  

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

• Flink 以其性能&易用性在實時計算領(lǐng)域開始大行其道，我們也在做一些業(yè)務(wù)試點;

• 實時和離線融合 CEP 場景的試水；

• Kafka 新版本的引入，包括 Qouta 限速、JBOD Markdown、Stream API、Excatly Once 等功能的支持；

• 實時平臺集成到統(tǒng)一的一體化平臺，用戶在一個平臺完成實時 & 離線開發(fā)；

• 發(fā)掘線上的業(yè)務(wù)場景，比如我們目前和策略部門合作的實時營銷項目就是通過用戶的行為數(shù)據(jù)來做一些策略，提高轉(zhuǎn)化率。

最后說一下關(guān)于平臺化演進中的心得：

• 學(xué)會資源聚集（善于借助外力，must to have VS nice to have）；

• MVP 最小化可行性調(diào)研（產(chǎn)品是迭代出來的，不是一蹴而就的，先完成后完美）；

• 偷懶思想，不重復(fù)造輪子（他山之石可以攻玉）；

• 賦能用戶，盡量自動化&自助化（提高效率，解放生產(chǎn)力）；

• 數(shù)據(jù)化運營（用數(shù)據(jù)說話）；

• 資源隔離，重點業(yè)務(wù)的 SLA 保證（降級限流策略，反壓功能等）；

• 做好監(jiān)控（全鏈路的監(jiān)控，并做必要測試）；

• 防范墨菲定律，及時做好容量規(guī)劃 & 壓測，同時不斷完善 SOP；

• 抽象思維（從一個到一類問題的抽象）；

• 以解決實際問題為目的，而不是炫技；

• 關(guān)注用戶體驗（做完了 VS 做好了）；

• 防火勝于救火，多想幾步，不斷總結(jié)并完善標準 & 規(guī)劃 & 流程（上線規(guī)范/ Checklist / SOP 流程等）。

近期熱文

云象區(qū)塊鏈CEO黃步添 | 分布式賬本

《白話區(qū)塊鏈》蔣勇 | 白話區(qū)塊鏈技術(shù)棧與應(yīng)用

元界CTO陳浩 | 從區(qū)塊鏈即服務(wù)（BaaS）到價值互聯(lián)網(wǎng)