導讀：微軟的ASG (應用與服務集團）包含Bing,、Office,、Skype。每天產(chǎn)生多達5 PB以上數(shù)據(jù)，如何構(gòu)建一個高擴展性的data audit服務來保證這樣量級的數(shù)據(jù)完整性和實時性非常具有挑戰(zhàn)性。本文將介紹微軟ASG大數(shù)據(jù)團隊如何利用Kafka、Spark以及Elasticsearch來解決這個問題。

案例簡介 

本案例介紹了微軟大數(shù)據(jù)平臺團隊設計和部署的基于開源技術(shù)（Kafka、Spark、ElasticsSearch、Kibana）的大數(shù)據(jù)質(zhì)量監(jiān)控平臺，這個平臺具有實時、高可用、可擴展、高度可信的特性，成為微軟Bing、Office365、Skype等年收入270+億美元的業(yè)務在監(jiān)控數(shù)據(jù)質(zhì)量方面的可靠技術(shù)保障。

同時，基于業(yè)務需要，我們在設計和實現(xiàn)中達成下面一系列的目標：

監(jiān)控流式數(shù)據(jù)的完整性與時延；

需要監(jiān)控的數(shù)據(jù)管道(pipeline)具有多個數(shù)據(jù)生產(chǎn)者、多處理階段、多數(shù)據(jù)消費者的特性；

數(shù)據(jù)質(zhì)量的監(jiān)控需要近實時(near real time)；

數(shù)據(jù)質(zhì)量發(fā)生問題的時候，需要提供相應的診斷信息來幫助工程師迅速解決問題；

監(jiān)控平臺的服務本身需要超級穩(wěn)定和高可用, 大于99.9%在線時間；

監(jiān)控與審計本身是高度可信；

平臺架構(gòu)可以水平擴展 (Scale out)。

背景及問題引入 

為了服務微軟的Bing、Office 365以及Skype業(yè)務，我們的大數(shù)據(jù)平臺需要處理每天高達十幾PB級別的海量大數(shù)據(jù)，所有的數(shù)據(jù)分析、報表、洞見以及A/B測試都依賴于高質(zhì)量的數(shù)據(jù)，如果數(shù)據(jù)質(zhì)量不高的話，依賴數(shù)據(jù)做決策的業(yè)務都會受到嚴重影響。

與此同時，微軟業(yè)務對于實時數(shù)據(jù)處理的需求也日益增加，以前監(jiān)控批處理數(shù)據(jù)（batch data）的很多解決方案已經(jīng)不再適用于實時的流式數(shù)據(jù)的質(zhì)量監(jiān)控。

在另外一個層面，基于歷史原因，各個業(yè)務集團往往使用不同的技術(shù)、工具來做數(shù)據(jù)處理，怎么整合這樣異構(gòu)的技術(shù)、工具以及在此之上的數(shù)據(jù)質(zhì)量監(jiān)控也是一個急需解決的問題。

圖1是我們數(shù)據(jù)處理平臺的一個概念性架構(gòu)。從數(shù)據(jù)生產(chǎn)者這端，我們通過在客戶端以及服務端使用通用的SDK，按照通用的schema來產(chǎn)生數(shù)據(jù)，數(shù)據(jù)通過分布在全世界的數(shù)據(jù)收集服務(collectors)來分發(fā)到相應的Kafka, 然后通過pub/sub模式由各種各樣的計算以及存儲框架來訂閱。

這樣各種團隊就可以選擇他們最熟悉或者一直以來使用的工具來做處理。例如，從實時處理的角度，各個業(yè)務團隊可以選用比如Spark或者微軟的USQL streaming處理框架，以及其他第三方的工具來做一些特定場景的分析，比如日志分析的Splunk、交互式分析的Interana等。在批處理框架上，用戶可以選用開源社區(qū)的Hadoop,、Spark或者微軟的Cosmos等。

圖1： 整合各個業(yè)務集團的異構(gòu)數(shù)據(jù)系統(tǒng)的架構(gòu)

圖2：快速增長的實時數(shù)據(jù)

如圖2所示，我們在遷移大數(shù)據(jù)到圖1架構(gòu)的過程中，也看到實時流式數(shù)據(jù)的快速增長。每天峰值消息高達一萬億個以上，每秒處理一百三十萬個消息， 每天處理3.5PB流式數(shù)據(jù)。

數(shù)據(jù)監(jiān)控的場景以及工作原理 

3.1數(shù)據(jù)監(jiān)控場景

基于業(yè)務需求，我們總結(jié)概括了需要被監(jiān)控的數(shù)據(jù)處理管道特性（如圖3）

多數(shù)據(jù)生產(chǎn)者（multiple data producers），數(shù)據(jù)來自客戶端和服務端；

多個數(shù)據(jù)消費者(multiple data consumers)，這里特指各種數(shù)據(jù)處理框架；

多數(shù)據(jù)監(jiān)控階段（multiple stages），從數(shù)據(jù)產(chǎn)生到數(shù)據(jù)處理，數(shù)據(jù)往往流經(jīng)多個數(shù)據(jù)管道的組件，我們需要通過監(jiān)控確保每個階段數(shù)據(jù)都不會發(fā)生丟失、高時延、以及異常。

圖3： 多數(shù)據(jù)生產(chǎn)者、多階段、多數(shù)據(jù)消費者的數(shù)據(jù)管道

3.2工作原理

基于圖3的數(shù)據(jù)管道，我們把問題具體化為如何確?；贙afka的數(shù)據(jù)管道上下游的數(shù)據(jù)完整性、實時性、數(shù)據(jù)異常的監(jiān)測。圖4是一個抽象化的監(jiān)控架構(gòu)以及工作原理。

藍色組件是數(shù)據(jù)管道里數(shù)據(jù)流經(jīng)的各個處理階段；綠色組件是本文中實時數(shù)據(jù)質(zhì)量監(jiān)控的核心服務Audit Trail。在數(shù)據(jù)流經(jīng)各個組件的同時，相應的審計(audit)數(shù)據(jù)也會同時發(fā)到Audit Trail, 這個審計數(shù)據(jù)可以看作是一種元數(shù)據(jù)(meta data)，它包含關(guān)于數(shù)據(jù)流的信息，例如該消息是在哪個數(shù)據(jù)中心、哪臺機器產(chǎn)生；該消息包含幾條記錄、大小、時間戳等。Audit Trail匯總了各個數(shù)據(jù)處理組件發(fā)來的元數(shù)據(jù)后，就可以實時做各種數(shù)據(jù)質(zhì)量的評估，比如數(shù)據(jù)在此時刻的完整性如何、實時性如何、有無異常。

圖4：數(shù)據(jù)流與監(jiān)控流，監(jiān)控流實時匯總到Audit Trail

基于圖5的審計元數(shù)據(jù)，一旦發(fā)生數(shù)據(jù)質(zhì)量問題，工程師可以快速定位是哪個數(shù)據(jù)中心的哪臺服務器在什么時間段發(fā)生了問題，然后快速采取相應行動來解決或緩解問題，并把對下游數(shù)據(jù)處理的影響降到較低。

圖5： 審計元數(shù)據(jù)的結(jié)構(gòu)

可被監(jiān)控的數(shù)據(jù)質(zhì)量問題可以分為如下幾類：

數(shù)據(jù)時延超出規(guī)定的SLA (service level agreement)

工程師可以通過如圖6所示的時延狀態(tài)圖快速了解在數(shù)據(jù)質(zhì)量時延這個維度是否正常，這對于對實時性要求比較嚴格的數(shù)據(jù)產(chǎn)品及應用非常重要，如果數(shù)據(jù)延遲到來，很多時候就失去了意義。

需要注意的是，圖表在這里起到的只是輔助作用，在真正的生產(chǎn)環(huán)境中是通過系統(tǒng)API調(diào)用來定期檢查SLA的符合情況，一旦超出時延閾值，會通過電話、短信等手段通知值班的工程師來實時解決問題。

圖6：簡單時延柱狀圖

數(shù)據(jù)在移動中發(fā)生丟失導致完整性不滿足SLA (service level agreement)

工程師可以通過圖7中所示簡單圖表來了解數(shù)據(jù)完整性的狀態(tài)，圖7所示包含兩個數(shù)據(jù)處理階段：一個數(shù)據(jù)生產(chǎn)者和兩個數(shù)據(jù)消費者的應用案例。所以圖表中實際上是三條線，綠色是生產(chǎn)者的實時數(shù)據(jù)量，藍色和紫色線是兩個數(shù)據(jù)消費者處理的數(shù)據(jù)量。如果在理想情況下，數(shù)據(jù)完整性沒有問題，這三條線是完全重合。本例中在最后一個點出現(xiàn)了分叉，代表數(shù)據(jù)完整性出現(xiàn)問題，需要工程師進行干預。

圖7：簡單完整性圖表

數(shù)據(jù)本身發(fā)生異常-通過異常檢測來實時監(jiān)控

數(shù)據(jù)本身發(fā)生異常，我們由相應的基于統(tǒng)計元數(shù)據(jù)的異常檢測（如圖8）來做實時監(jiān)控。異常檢測是一個在工業(yè)界非常普遍的問題和挑戰(zhàn)，幾乎每個互聯(lián)網(wǎng)公司都會有做異常檢測的服務或平臺，但是做好很不容易，這是一個可以單獨寫一篇文章的大題目，這里只是單辟一個章節(jié)做簡單的算法介紹。

圖8：基于審計數(shù)據(jù)的異常檢測

本例是通過對于數(shù)據(jù)量的異常檢測來發(fā)現(xiàn)上游寫log問題，或者其他數(shù)據(jù)生產(chǎn)的邏輯問題。

3.3異常檢測

異常檢測算法1

圖 9 Holt-Winters算法

我們采用了Holt-Winters算法（圖9）來訓練模型和做預測，并在此之上做了很多改進來增加算法的強健性和容錯能力。

強健性上的改進包括：

使用Median Absolute Deviation (MAD) 得到更好的估值；

處理數(shù)據(jù)丟點和噪聲 (例如數(shù)據(jù)平滑)。

功能上的改進包括：

自動獲取趨勢和周期信息；

允許用戶人工標記和反饋來更好的處理趨勢變化。

通過比較預測值和實際值，我們采用GLR (Generalized Likelihood Ratio) 來發(fā)現(xiàn)異常點。在這上面我們也做了相應的改進，包括：

Floating Threshold GLR, 基于新的輸入數(shù)據(jù)動態(tài)調(diào)整模型；

對于噪聲比較大的數(shù)據(jù)做去除異常點。

異常檢測算法2

這是一個基于Exchangeability Martingale的在線時間序列的異常檢測算法，其核心就是假設數(shù)據(jù)的分布是穩(wěn)定的。如果新的數(shù)據(jù)點的加入導致數(shù)據(jù)的分布(distribution)發(fā)生比較大的變化，我們就認為異常發(fā)生了。所以基于歷史數(shù)據(jù)，我們需要定義一個新值異常公式（New value strangeness）。下面是這些公式的構(gòu)成，對數(shù)學不感興趣的讀者可以略去。

在某個時刻t, 我們收到一個新的數(shù)據(jù)點，對于歷史每個數(shù)據(jù)i：

s[i] = strangeness function of (value[i], history)

Let  p[t] = (#{i: s[i] > s[t]}+ r*#{i: s[i]==s[t]})/N, where r is uniform in (0,1)

Uniform r makes sure p is uniform

Exchangeability Martingale: Mt=i=1t?pi?-1

EMtp1,p2,…pt-1=Mt-1

Integrate ?pi?-1 over [0,1] and pi is uniform

報警觸發(fā)門檻通過Doob’s maximal inequality控制

Prob (? t :Mt>λ)<1λ

對于異常點，Martingale的值就會大于門檻值。

異常檢測算法3

這是一個簡單而非常有效的基于歷史數(shù)據(jù)的指數(shù)平滑算法。

它首先基于歷史數(shù)據(jù)生成動態(tài)上下界：

Threshold (width) = min(max(M1*Mean, M2*Standard Deviation), M3*Mean)   (M1<M3)

Alert: |Value – predicated value| > Threshold

預測值 = S1+12S2+14S3+18S4+116S51+12+14+18+116

優(yōu)點在于處理周期性數(shù)據(jù)的異常檢測很好，并且允許用戶反饋和標記來調(diào)整動態(tài)上下界。

系統(tǒng)設計概述 

基于業(yè)務場景的需要，我們在設計和實現(xiàn)中需要達成一系列的目標以及處理相應的挑戰(zhàn)：

監(jiān)控流式數(shù)據(jù)的完整性與時延；

需要監(jiān)控的數(shù)據(jù)管道(pipeline)具有多個數(shù)據(jù)生產(chǎn)者、多處理階段、多數(shù)據(jù)消費者的特性；

數(shù)據(jù)質(zhì)量的監(jiān)控需要近實時(near real time)；

數(shù)據(jù)發(fā)生問題的時候，提供相應的診斷信息來幫助工程師迅速解決問題；

監(jiān)控平臺的服務本身需要超級穩(wěn)定和高可用, 99.9%以上在線時間；

監(jiān)控與審計本身是高度可信；

平臺架構(gòu)可以水平擴展 (Scale out)。

4.1高可用可擴展的架構(gòu)

如圖10所示，審計元數(shù)據(jù)通過前端服務（front end web service）到達Kafka, 我們利用Kafka來實現(xiàn)高可用的臨時存儲（transient storage）, 這樣，我們的數(shù)據(jù)生產(chǎn)者和消費者在發(fā)送審計數(shù)據(jù)的同時，就不會發(fā)生阻塞進而影響更重要的數(shù)據(jù)流。

通過Spark streaming的應用，把審計數(shù)據(jù)按照時間窗口聚合，同時有相應的邏輯處理去重，晚到以及非順序到來的數(shù)據(jù)，同時做各種容錯處理保證高可用。

ElasticsSearch作為存儲聚合的審計數(shù)據(jù)，通過Kibana做報表展示，進而通過Data Analysis service對外提供API來使得用戶獲取各種數(shù)據(jù)質(zhì)量信息。

Data Analysis Service作為最終的API端，提供各種數(shù)據(jù)完整性、實時性、異常的信息。

上述組件，每個都設計成可以獨立水平擴展(Scale out), 并且在設計上保證高容錯已實現(xiàn)高可用性。

圖10：Audit Trail數(shù)據(jù)處理架構(gòu)

4.2異地雙活的可靠性保障

通過雙數(shù)據(jù)中心Active-Active災備（Disaster recovery）如圖11所示，來進一步保證高可用高可靠的服務。整體架構(gòu)保證數(shù)據(jù)流同時通過兩個同構(gòu)的審計處理管道進行處理，即使一個數(shù)據(jù)中心因為各種原因下線，整體服務還是處于可用狀態(tài)，進而保證全天候的數(shù)據(jù)質(zhì)量審計與監(jiān)控。

圖11：雙數(shù)據(jù)中心Active-Active Disaster Recovery

4.3高度可信的審計與監(jiān)控服務

對于任何監(jiān)控服務來說，經(jīng)常被質(zhì)疑的就是是否監(jiān)控服務本身的結(jié)果是準確可信的。為了保證這一點，我們通過兩種方式來保證服務的可信度：

用來審計自身（Audit for audit）（圖12）；

Synthetic probe。

圖12：審計自身

在基于Kafka/Spark/ES的管道之外，我們還有一套獨立的經(jīng)由ES的審計元數(shù)據(jù)的處理管道，通過比較上述兩個管道的結(jié)果，我們就能保證審計數(shù)據(jù)的可靠性。

另外，基于synthetic probe的方式，我們每分鐘會發(fā)送一組synthetic數(shù)據(jù)進入前端服務（front end web service）, 然后試圖從Data Analysis web service 讀出，通過這種方式進一步保障數(shù)據(jù)的可靠性。

4.4輔助數(shù)據(jù)質(zhì)量問題的診斷

當數(shù)據(jù)質(zhì)量發(fā)生問題，Audit Trail提供了原始的審計元數(shù)據(jù)來幫助工程師進一步做問題的診斷。工程師可以使用這些元數(shù)據(jù)和他們自己的trace來進一步JOIN, 來提供一種交互式的診斷，如圖13。

圖13：把Trace和審計元數(shù)據(jù)做JOIN, 可視化的交互診斷視圖

效果評估與總結(jié) 

通過上述系統(tǒng)架構(gòu)的設計與部署，我們實現(xiàn)了一系列支持公司Bing,、Office,、Skype業(yè)務發(fā)展的數(shù)據(jù)質(zhì)量監(jiān)控目標：

監(jiān)控流式數(shù)據(jù)的完整性與時延；

需要監(jiān)控的數(shù)據(jù)管道(pipeline)具有多個數(shù)據(jù)生產(chǎn)者、多處理階段、多數(shù)據(jù)消費者的特性；

數(shù)據(jù)質(zhì)量的監(jiān)控需要近實時(near real time)；

數(shù)據(jù)發(fā)生問題的時候，需要提供相應的診斷信息來幫助工程師迅速解決問題；

監(jiān)控平臺的服務本身需要超級穩(wěn)定和高可用, 99.9%在線時間

監(jiān)控與審計本身是高度可信；

平臺架構(gòu)可以水平擴展 (Scale out)。

同時，我們準備開源這個平臺服務，因為我們相信這個服務本身是一個足夠通用化的解決方案，可以應用于很多公司的數(shù)據(jù)質(zhì)量監(jiān)控場景。