揭祕|每秒千萬級的實時數據處理是怎麼實現的？

一、設計背景

閒魚目前實際生產部署環境愈來愈複雜，橫向依賴各類服務盤宗錯節，縱向依賴的運行環境也愈來愈複雜。當服務出現問題的時候，可否及時在海量的數據中定位到問題根因，成爲考驗閒魚服務能力的一個嚴峻挑戰。

線上出現問題時經常須要十多分鐘，甚至更長時間才能找到問題緣由，所以一個可以快速進行自動診斷的系統需求就應用而生，而快速診斷的基礎是一個高性能的實時數據處理系統。

這個實時數據處理系統須要具有以下的能力：

一、數據實時採集、實時分析、複雜計算、分析結果持久化。
二、能夠處理多種多樣的數據。包含應用日誌、主機性能監控指標、調用鏈路圖。
三、高可靠性。系統不出問題且數據不能丟。
四、高性能，底延時。數據處理的延時不超過3秒，支持每秒千萬級的數據處理。

本文不涉及問題自動診斷的具體分析模型，只討論總體實時數據處理鏈路的設計。

二、輸入輸出定義

爲了便於理解系統的運轉，咱們定義該系統總體輸入和輸出以下：

 輸入：

服務請求日誌（包含traceid、時間戳、客戶端ip、服務端ip、耗時、返回碼、服務名、方法名）
       
環境監控數據（指標名稱、ip、時間戳、指標值）。好比cpu、 jvm gc次數、jvm gc耗時、數據庫指標。

輸出：

一段時間內的某個服務出現錯誤的根因，每一個服務的錯誤分析結果用一張有向無環圖表達。（根節點便是被分析的錯誤節點，葉子節點便是錯誤根因節點。葉子節點多是一個外部依賴的服務錯誤也多是jvm異常等等）。

三、架構設計

 在實際的系統運行過程當中，隨着時間的推移，日誌數據以及監控數據是源源不斷的在產生的。每條產生的數據都有一個本身的時間戳。而實時傳輸這些帶有時間戳的數據就像水在不一樣的管道中流動同樣。

若是把源源不斷的實時數據比做流水，那數據處理過程和自來水生產的過程也是相似的：

天然地，咱們也將實時數據的處理過程分解成採集、傳輸、預處理、計算、存儲幾個階段。

總體的系統架構設計以下：

採集

採用阿里自研的sls日誌服務產品（包含logtail+loghub組件），logtail是採集客戶端，之因此選擇logtail是由於其優秀的性能、高可靠性以及其靈活插件擴展機制，閒魚能夠定製本身的採集插件實現各類各樣數據的實時採集。

傳輸

loghub能夠理解爲一個數據發佈訂閱組件，和kafka的功能相似，做爲一個數據傳輸通道其更穩定、更安全，詳細對比文章參考：https://yq.aliyun.com/articles/35979?spm=5176.10695662.1996646101.searchclickresult.6f2c7fbe6g3xgP

預處理

實時數據預處理部分採用blink流計算處理組件（開源版本叫作flink，blink是阿里在flink基礎上的內部加強版本）。目前經常使用的實時流計算開源產品有Jstorm、SparkStream、Flink。Jstorm因爲沒有中間計算狀態的，其計算過程當中須要的中間結果必然依賴於外部存儲，這樣會致使頻繁的io影響其性能；SparkStream本質上是用微小的批處理來模擬實時計算，實際上仍是有必定延時；Flink因爲其出色的狀態管理機制保證其計算的性能以及實時性，同時提供了完備SQL表達，使得流計算更容易。

 計算與持久化

數據通過預處理後最終生成調用鏈路聚合日誌和主機監控數據，其中主機監控數據會獨立存儲在tsdb時序數據庫中，供後續統計分析。tsdb因爲其針對時間指標數據的特別存儲結構設計，很是適合作時序數據的存儲與查詢。調用鏈路日誌聚合數據，提供給cep/graph service作診斷模型分析。cep/graph service是閒魚自研的一個應用，實現模型分析、複雜的數據處理以及外部服務進行交互，同時藉助rdb實現圖數據的實時聚合。

最後cep/graph service分析的結果做爲一個圖數據，實時轉儲在lindorm中提供在線查詢。lindorm能夠看做是加強版的hbase，在系統中充當持久化存儲的角色。

四、設計細節與性能優化

採集

日誌和指標數據採集使用logtail，整個數據採集過程如圖：

其提供了很是靈活的插件機制，共有四種類型的插件：

• inputs: 輸入插件，獲取數據。
• processors: 處理插件，對獲得的數據進行處理。
• aggregators: 聚合插件，對數據進行聚合。
• flushers: 輸出插件，將數據輸出到指定 sink。

因爲指標數據（好比cpu、內存、jvm指標）的獲取須要調用本地機器上的服務接口獲取，所以應儘可能減小請求次數，在logtail中，一個input佔用一個goroutine。閒魚經過定製input插件和processors插件，將多個指標數據（好比cpu、內存、jvm指標）在一個input插件中經過一次服務請求獲取（指標獲取接口由基礎監控團隊提供），並將其格式化成一個json數組對象，在processors插件中再拆分紅多條數據，以減小系統的io次數同時提高性能。

傳輸

數據傳輸使用LogHub，logtail寫入數據後直接由blink消費其中的數據，只需設置合理的分區數量便可。分區數要大於等於bink讀取任務的併發數，避免blink中的任務空轉。

預處理

預處理主要採用bink實現，主要的設計和優化點：

1:編寫高效的計算流程

blink是一個有狀態的流計算框架，很是適合作實時聚合、join等操做。
在咱們的應用中只須要關注出現錯誤的的請求上相關服務鏈路的調用狀況，所以整個日誌處理流分紅兩個流：
a、服務的請求入口日誌做爲一個單獨的流來處理，篩選出請求出錯的數據。
b、其餘中間鏈路的調用日誌做爲另外一個獨立的流來處理，經過和上面的流join on traceid實現出錯服務依賴的請求數據塞選。

 如上圖所示經過雙流join後，輸出的就是全部發生請求錯誤相關鏈路的完整數據。

2:設置合理的state生存週期

blink在作join的時候本質上是經過state緩存中間數據狀態，而後作數據的匹配。而若是state的生命週期太長會致使數據膨脹影響性能，若是state的生命週期過短就會沒法正常關聯出部分延遲到來的數據，因此須要合理的配置state生存週期，對於該應用容許最大數據延遲爲1分鐘。

使用niagara做爲statebackend，以及設定state數據生命週期，單位毫秒
state.backend.type=niagara
state.backend.niagara.ttl.ms=60000

3:開啓 MicroBatch/MiniBatch

MicroBatch 和 MiniBatch 都是微批處理，只是微批的觸發機制上略有不一樣。原理上都是緩存必定的數據後再觸發處理，以減小對 state 的訪問從而顯著提高吞吐，以及減小輸出數據量。

開啓join
blink.miniBatch.join.enabled=true
使用 microbatch 時須要保留如下兩個 minibatch 配置
blink.miniBatch.allowLatencyMs=5000
防止OOM，每一個批次最多緩存多少條數據
blink.miniBatch.size=20000

4:動態負載使用 Dynamic-Rebalance 替代 Rebalance

blink任務在運行是最忌諱的就是存在計算熱點，爲保證數據均勻使用Dynamic Rebalance，它能夠根據當前各subpartition中堆積的buffer的數量，選擇負載較輕的subpartition進行寫入，從而實現動態的負載均衡。相比於靜態的rebalance策略，在下游各任務計算能力不均衡時，可使各任務相對負載更加均衡，從而提升整個做業的性能。

開啓動態負載
task.dynamic.rebalance.enabled=true

5:自定義輸出插件

數據關聯後須要將統一請求鏈路上的數據做爲一個數據包通知下游圖分析節點，傳統的方式的是經過消息服務來投遞數據。可是經過消息服務有兩個缺點：
一、其吞吐量和rdb這種內存數據庫相比比仍是較大差距（大概差一個數量級）。
二、在接受端還須要根據traceid作數據關聯。

咱們經過自定義插件的方式將數據經過異步的方式寫入RDB，同時設定數據過時時間。在RDB中以數據結構存儲。寫入的同時只將traceid作爲消息內容經過metaQ通知下游計算服務，極大的減小了metaQ的數據傳輸壓力。

圖聚合計算

cep/graph計算服務節點在接收到metaQ的通知後，綜合根據請求的鏈路數據以及依賴的環境監控數據，會實時生成診斷結果。診斷結果簡化爲以下形式：

說明本次請求是因爲下游jvm的線程池滿致使的，可是一次調用並不能說明該服務不可用的根本緣由，須要分析總體的錯誤狀況，那就須要對圖數據作實時聚合。

聚合設計以下（爲了說明基本思路，作了簡化處理）：

一、首先利用redis的zrank能力爲根據服務名或ip信息爲每一個節點分配一個全局惟一排序序號。
二、爲圖中的每一個節點生成對應圖節點編碼，編碼格式：
            對於頭節點：頭節點序號|歸整時間戳|節點編碼
            對於普通節點：|歸整時間戳|節點編碼
三、因爲每一個節點在一個時間週期內都有惟一的key，所以能夠將節點編碼做爲key利用redis爲每一個節點作計數。同時消除了併發讀寫的問題。
四、利用redis中的set集合能夠很方便的疊加圖的邊。
五、記錄根節點，便可經過遍歷還原聚合後的圖結構。

聚合後的結果大體以下：
                     

這樣最終生成了服務不可用的總體緣由，而且經過葉子節點的計數能夠實現根因的排序。

五、收益

系統上線後，整個實時處理數據鏈路的延遲不超過三秒。閒魚服務端問題的定位時間從十多分鐘甚至更長時間降低到五秒內。大大的提高了問題定位的效率。

六、將來展望

目前的系統能夠支持閒魚每秒千萬的數據處理能力。後續自動定位問題的服務可能會推廣到阿里內部更多的業務場景，隨之而來的是數據量的成倍增長，所以對於效率和成本提出了更好的要求。

將來咱們可能作的改進：

一、可以自動的減小或者壓縮處理的數據。
二、複雜的模型分析計算也能夠在blink中完成，減小io，提高性能。
三、支持多租戶的數據隔離。

---

本文做者：閒魚技術-靖楊

閱讀原文

本文爲雲棲社區原創內容，未經容許不得轉載。