夏軍：小米大數據集成架構演化之路

【IT168 專稿】本文根據夏軍老師在2018年10月18日【第十屆中國系統架構師大會】現場演講內容整理而成。

講師簡介：

夏軍，小米數據流平台負責人，曾就職於騰訊和百度，主要負責消息隊列、大數據集成方案，離線計算和實時計算等方面的工作。

摘要：

小米有眾多的智能終端和設備，數據規模非常大，對於數據採集和大數據集成提出了非常高的要求。此次演講主要介紹小米大數據集成解決方案，主要包括小米數據流平台的架構演化，整個鏈路的數據質量監控，數據流生態的構建思路，最後會介紹典型的應用場景、未來的規劃和思考。

分享大綱：

1、問題與挑戰

2、數據流整體框架

3、核心功能

4、應用場景解析

正文：

1、問題與挑戰

首先，我介紹一下小米大數據集成架構面臨的問題和挑戰：一是大數據場景下系統眾多，包括各種存儲系統和計算系統，這其中很關鍵的一個問題是如何高效集成所有系統以讓數據發揮最大價值;二是做數據集成時，我們希望可以有更低延遲;三是當數據在不同系統中流動時，我們如何及時發現並解決問題;四是量化數據在整個鏈路傳輸過程中存在的問題，比如數據延遲、數據丟失等。

2、數據流整體框架

上圖為小米數據流整體架構，大致可分為三部分：中間層叫Talos，這是小米自研的一套消息隊列，其主要應用場景有兩個：一是作為數據中間件(數據中轉中心)，二是服務於後續的流式計算。雖然現在Kafka非常火，但是我們確實發現了Kafka的一些問題，比如Reblance、擴容、縮容等問題，因此我們選擇使用自研Talos。下層是基於流式消息隊列做的source和sink擴展，目標是希望以Talos為數據匯流排把大數據應用場景下的不同平台連接起來。上層依賴於底層的source和sink體系，解決Metric監控、報警和數據收集等問題，也會做OLAP分析和線上APP日誌收集等。我們希望在這個架構下，業務方可以根據不同需求適配該系統以得到完美的解決方案。

3、核心功能

上圖為該平台的主要核心模塊，最底層是消息隊列，中間層是數據接入層，該層包含一些SDK，因為它本身作為消息隊列也存在很多應用場景，比如推送場景等。我們在消息隊列上做了Streaming Plunigs以適配各種Streaming系統，其後的Source和Sink其實是對數據的擴展。最上層主要是基於這套框架做的整體feature，比如全局web端控制和產品化方案。我們也做了全鏈路數據監控和數據追蹤。我們有一個自己的流式計算管理平台，用來幫助管理用戶流式作業。因為小米有自己的海外業務，所以我們有全局數據中心的數據replication機制，需要在海外部署自己的數據中心，因為數據是全球分散化的，必然就存在數據匯總問題，這些構成了我們的整體解決方案。

接下來，我將逐步介紹核心功能。首先，我介紹一般數據解決方案(如上圖)，這是一些常見系統，我們要解決的問題是在不同的系統間做數據集成。

這之中存在一些問題：一是系統的交互複雜度較高;二是當涉及的系統較多時，如果每個業務都做自己的事情，不僅前期研發成本會非常高，而且後續功能添加、系統運維、系統重構和交接成本會更高;三是當各業務方按照自身需求進行開發時，由於彼此獨立導致無法復用重複邏輯;四是如果讓業務方完成某件事情，業務方往往會忽略監控和數據質量，一般缺失或者很難做到很完備，因此無法保證數據交互質量;五是一般由業務獨立部署，基本無法抽象化和服務化，進而很難積累經驗和傳承知識。

我們的做法是基於Talos消息隊列作為消息匯流排，和不同的系統進行交互，我們在外圍做一些產品化工作，接管一些比較繁瑣的配置。基於此，我們做了Multi Source和Multi Sink的抽象。

我們認為抽象首先是希望數據在系統間進行流動，這之中包含兩層含義：一是連接不同的系統;二是構建低延遲場景。我們做產品化封裝其實就是避免業務團隊的重複投入與研發。我們希望以Talos為消息數據匯流排對所有數據使用流式計算以儘可能降低數據結構產生的延遲。要想在所有系統之間作中轉，我們就需要考慮集成複雜度，通過Source和Sink組合的模式，系統集成複雜度降為O(N)。最後，我們期望按照這種模式接入更多系統，這樣我們就可以形成規模效應，依賴目前的數據流平台給用戶產生價值，建立數據流生態系統。

接下來，我們介紹一下系統監控。對於整套流程，我認為系統監控大概分為以下幾方面：數據丟失監控，數據延遲增加監控，服務進程異常監控，流量異常監控。

一般情況下，我們會在每台機器上部署一個agent，收集不同模塊的Metric數據。其次，我們會把數據匯總到消息隊列，對數據進行各種整理並中轉到Druid平台，由於監控數據有很多維度，因此我們用Druid的目的就是降維，按照不同維度進行數據合併。然後，我們將數據中轉到Falcon，這是一套小米開源的監控系統，我們會對監控數據提供Web化展示，讓用戶實時看到監控內容，我們也會周期性生成報表來告訴用戶一些數據情況。

說到底，數據流端到端審計其實就是量化整個鏈路的數據情況，其展現形式就是Web界面，能夠實時查詢最新數據，也能查詢歷史數據。在報表部分，我們引入了兩個概念：Event Time和 Processing Time，Event Time指消息真正產生的系統時間。Processing Time指消息到達具體某個模塊，因為我們是一個流動的系統，裡面有很多模塊，我們會把 Event Time和 Processing Time依賴於類似Metric的模塊進行數據打點並收集。我們會把Event Time看作消息的唯一標籤來進行數據校驗。Processing Time用來統計消息從產生到達某一模塊的系統延遲情況。

上圖為數據流端到端審計的簡單處理流程，我們會在每一個節點做埋點數據，埋點數據主要包括每條消息的Event Time和 Processing Time，以及其歸屬的流等類似信息。為了保證監控數據的高可用，我們會把監控數據先持久化到本地磁碟。然後，我們通過一套自己的流程收集數據。

整個過程存在幾個問題，一是監控數據量非常大，如果要監控整個鏈路，我們需要給每條流經系統的數據做埋點，這會導致系統負載增加;二是消息有可能重複，在最終結果匯總時，我們要做相應的去重處理;三是期望數據以一種准實時的形式展示給用戶。

在這之中，我們引入了Spark Streaming對數據進行處理，雖然消息有重複，雖然Spark Streaming作業會因為各種原因掛掉，但我們需要保證消息正確統計。我們在Agent端進行分鐘級別的數據合併，也就是在每台機器上進行預聚合，這樣能夠把原始監控消息的數據量大大降低。我們會在Spark Streaming里做一些基於內存的去重邏輯，依賴外部KV系統做數據校驗。

上圖為我們的監控成果，小米會有很多線上數據，這些數據在進行埋點後寫入消息隊列，最後會轉到Kudu和HDFS中，HDFS後期會做一些基於離線的Hive分析過渡，或者是OLAP分析。上圖展示了我們到達這個環節的數據量，對於一個時間窗口，也就是一分鐘的時間內，窗口所到達的數據量。

上圖顯示出現了一些數據丟失。目前來看，整個過程應該是數據從線上灌入Hive和HDFS，上圖所示數據為測試樣例，展示了從消息隊列到後端部分系統做轉儲的過程。

4、應用場景解析

因為篇幅有限，我就介紹了小米對Source和Sink理念的理解，以及我們如何做監控和數據審計。我認為，對大數據集成系統而言，這三點對對業務來說是至關重要的。接下來，我將解析部分應用場景。

首先介紹小米內部數據集成的典型方案——埋點數據收集。對互聯網公司而言，埋點數據收集是非常重要的應用場景。上圖大概分為幾部分，一是對各種web服務設置埋點數據，通過擴展appender來實現業務，方便的把數據通過每台機器部署的agent進行中轉;其次，我們也做了web接入層，將大量的離散點通過這種模式收集起來，進而對其做數據分析。

第二個應用場景是實時日誌分析，對於線上日誌文件，我們會通過agent進行監控，將數據實時傳入Talos，通過ES或者Kibana進行實時查詢。簡單來說，我們依賴整個系統把分散式文件通過准實時的方式導入ES，再通過Kibana進行查詢和可視化日誌分析。

第三個場景是泛OLAP場景。我們通過Druid做多維度分析，使用Kudu進行即即席詢，利用Kudu的列存儲以較低延遲展示數據， 利用Kylin做T+1查詢。

整體流程如上圖所示，小米目前還存在大量MySQL集群和報表數據，MySQL可以滿足OTLP的需求，但是對於很多的OLAP查詢，延遲會比較大，無法滿足需求。基於此場景，我們的解決方案是把MySQL數據以准實時方式導出，在Kudu里做一個實時鏡像，通過Spark SQL做查詢。或者，將HDFS的數據灌入kudu，再接入Superset之類做展示。

最後一個場景是流式計算。我們之前提到系統中有很多Source和Sink，我們期望能夠從Talos也就是消息隊列里將數據傳入Spark Streaming做實時計算，綜上，這就是小米在數據流系統構建方面的經驗。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！