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

案例簡(jiǎn)介 

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

同時(shí)，基于業(yè)務(wù)需要，我們?cè)谠O(shè)計(jì)和實(shí)現(xiàn)中達(dá)成下面一系列的目標(biāo)：

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

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

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

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

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

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

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

背景及問(wèn)題引入 

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

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

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

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

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

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

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

如圖2所示，我們?cè)谶w移大數(shù)據(jù)到圖1架構(gòu)的過(guò)程中，也看到實(shí)時(shí)流式數(shù)據(jù)的快速增長(zhǎng)。每天峰值消息高達(dá)一萬(wàn)億個(gè)以上，每秒處理一百三十萬(wàn)個(gè)消息， 每天處理3.5PB流式數(shù)據(jù)。

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

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

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

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

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

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

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

3.2工作原理

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

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

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

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

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

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

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

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

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

圖6：簡(jiǎn)單時(shí)延柱狀圖

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

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

圖7：簡(jiǎn)單完整性圖表

數(shù)據(jù)本身發(fā)生異常-通過(guò)異常檢測(cè)來(lái)實(shí)時(shí)監(jiān)控

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

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

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

3.3異常檢測(cè)

異常檢測(cè)算法1

圖 9 Holt-Winters算法

我們采用了Holt-Winters算法（圖9）來(lái)訓(xùn)練模型和做預(yù)測(cè)，并在此之上做了很多改進(jìn)來(lái)增加算法的強(qiáng)健性和容錯(cuò)能力。

強(qiáng)健性上的改進(jìn)包括：

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

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

功能上的改進(jìn)包括：

自動(dòng)獲取趨勢(shì)和周期信息；

允許用戶人工標(biāo)記和反饋來(lái)更好的處理趨勢(shì)變化。

通過(guò)比較預(yù)測(cè)值和實(shí)際值，我們采用GLR (Generalized Likelihood Ratio) 來(lái)發(fā)現(xiàn)異常點(diǎn)。在這上面我們也做了相應(yīng)的改進(jìn)，包括：

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

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

異常檢測(cè)算法2

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

在某個(gè)時(shí)刻t, 我們收到一個(gè)新的數(shù)據(jù)點(diǎn)，對(duì)于歷史每個(gè)數(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

報(bào)警觸發(fā)門檻通過(guò)Doob’s maximal inequality控制

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

對(duì)于異常點(diǎn)，Martingale的值就會(huì)大于門檻值。

異常檢測(cè)算法3

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

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

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

Alert: |Value – predicated value| > Threshold

預(yù)測(cè)值 = S1+12S2+14S3+18S4+116S51+12+14+18+116

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

系統(tǒng)設(shè)計(jì)概述 

基于業(yè)務(wù)場(chǎng)景的需要，我們?cè)谠O(shè)計(jì)和實(shí)現(xiàn)中需要達(dá)成一系列的目標(biāo)以及處理相應(yīng)的挑戰(zhàn)：

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

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

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

數(shù)據(jù)發(fā)生問(wèn)題的時(shí)候，提供相應(yīng)的診斷信息來(lái)幫助工程師迅速解決問(wèn)題；

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

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

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

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

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

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

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

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

上述組件，每個(gè)都設(shè)計(jì)成可以獨(dú)立水平擴(kuò)展(Scale out), 并且在設(shè)計(jì)上保證高容錯(cuò)已實(shí)現(xiàn)高可用性。

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

4.2異地雙活的可靠性保障

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

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

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

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

用來(lái)審計(jì)自身（Audit for audit）（圖12）；

Synthetic probe。

圖12：審計(jì)自身

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

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

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

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

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

效果評(píng)估與總結(jié) 

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

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

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

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

數(shù)據(jù)發(fā)生問(wèn)題的時(shí)候，需要提供相應(yīng)的診斷信息來(lái)幫助工程師迅速解決問(wèn)題；

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

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

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

同時(shí)，我們準(zhǔn)備開源這個(gè)平臺(tái)服務(wù)，因?yàn)槲覀兿嘈胚@個(gè)服務(wù)本身是一個(gè)足夠通用化的解決方案，可以應(yīng)用于很多公司的數(shù)據(jù)質(zhì)量監(jiān)控場(chǎng)景。