本篇文章內容來自2016年TOP100summit Microsoft資深產品經理邢國冬的案例分享。
編輯：Cynthia

邢國冬（Tony Xing）：Microsoft資深產品經理、負責微軟應用與服務集團的大資料平臺構建，資料產品與服務.

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

一．案例簡介

本案例介紹了微軟大資料平臺團隊設計和部署的基於開源技術（Kafka、Spark、ElasticsSearch、Kibana）的大資料質量監控平臺，這個平臺具有實時、高可用、可擴充套件、高度可信的特性，成為微軟Bing、Office365、Skype等年收入270+億美元的業務在監控資料質量方面的可靠技術保障。同時，基於業務需要，我們在設計和實現中達成下面一系列的目標：

● 監控流式資料的完整性與時延；
● 需要監控的資料管道(pipeline)具有多個數據生產者、多處理階段、多資料消費者的特性；
● 資料質量的監控需要近實時(near real time)；
● 資料質量發生問題的時候，需要提供相應的診斷資訊來幫助工程師迅速解決問題；
● 監控平臺的服務本身需要超級穩定和高可用, 大於99.9%線上時間；
● 監控與審計本身是高度可信；
● 平臺架構可以水平擴充套件 (Scale out)。

二、背景以及問題的引入

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

與此同時，微軟業務對於實時資料處理的需求也日益增加，以前監控批處理資料（batch data）的很多解決方案已經不再適用於實時的流式資料的質量監控。

在另外一個層面，基於歷史原因，各個業務集團往往使用不同的技術、工具來做資料處理，怎麼整合這樣異構的技術、工具以及在此之上的資料質量監控也是一個急需解決的問題。

圖1是我們資料處理平臺的一個概念性架構。從資料生產者這端，我們通過在客戶端以及服務端使用通用的SDK，按照通用的schema來產生資料，資料通過分佈在全世界的資料收集服務(collectors)來分發到相應的Kafka,然後通過pub/sub模式由各種各樣的計算以及儲存框架來訂閱。

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

如圖2所示，我們在遷移大資料到圖1架構的過程中，也看到實時流式資料的快速增長。每天峰值訊息高達一萬億個以上，每秒處理一百三十萬個訊息， 每天處理3.5PB流式資料。

三、資料監控的場景以及工作原理

3.1資料監控場景

基於業務需求，我們總結概括了需要被監控的資料處理管道特性（如圖3）
● 多資料生產者（multiple data producers），資料來自客戶端和服務端；
● 多個數據消費者(multiple data consumers)，這裡特指各種資料處理框架；
● 多資料監控階段（multiple stages），從資料產生到資料處理，資料往往流經多個數據管道的元件，我們需要通過監控確保每個階段資料都不會發生丟失、高時延、以及異常。

3.2工作原理

基於圖3的資料管道，我們把問題具體化為如何確保基於Kafka的資料管道上下游的資料完整性、實時性、資料異常的監測。圖4是一個抽象化的監控架構以及工作原理。

藍色元件是資料管道里資料流經的各個處理階段；綠色元件是本文中實時資料質量監控的核心服務Audit Trail。在資料流經各個元件的同時，相應的審計(audit)資料也會同時發到Audit Trail, 這個審計資料可以看作是一種元資料(meta data)，它包含關於資料流的資訊，例如該訊息是在哪個資料中心、哪臺機器產生；該訊息包含幾條記錄、大小、時間戳等。Audit Trail彙總了各個資料處理元件發來的元資料後，就可以實時做各種資料質量的評估，比如資料在此時刻的完整性如何、實時性如何、有無異常。

基於圖5的審計元資料，一旦發生資料質量問題，工程師可以快速定位是哪個資料中心的哪臺伺服器在什麼時間段發生了問題，然後快速採取相應行動來解決或緩解問題，並把對下游資料處理的影響降到最低。

可被監控的資料質量問題可以分為如下幾類：
● 資料時延超出規定的SLA (service level agreement)

工程師可以通過如圖6所示的時延狀態圖快速瞭解在資料質量時延這個維度是否正常，這對於對實時性要求比較嚴格的資料產品及應用非常重要，如果資料延遲到來，很多時候就失去了意義。

需要注意的是，圖表在這裡起到的只是輔助作用，在真正的生產環境中是通過系統API呼叫來定期檢查SLA的符合情況，一旦超出時延閾值，會通過電話、簡訊等手段通知值班的工程師來實時解決問題。

● 資料在移動中發生丟失導致完整性不滿足SLA (service level agreement)

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

● 資料本身發生異常-通過異常檢測來實時監控

資料本身發生異常，我們由相應的基於統計元資料的異常檢測（如圖8）來做實時監控。異常檢測是一個在工業界非常普遍的問題和挑戰，幾乎每個網際網路公司都會有做異常檢測的服務或平臺，但是做好很不容易，這是一個可以單獨寫一篇文章的大題目，這裡只是單闢一個章節做簡單的演算法介紹。

本例是通過對於資料量的異常檢測來發現上游寫log問題，或者其他資料生產的邏輯問題。

3.3異常檢測

3.3.1異常檢測演算法1

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

強健性上的改進包括：
● 使用Median Absolute Deviation (MAD) 得到更好的估值；
● 處理資料丟點和噪聲 (例如資料平滑)。
功能上的改進包括：
● 自動獲取趨勢和週期資訊；
● 允許使用者人工標記和反饋來更好的處理趨勢變化。
通過比較預測值和實際值，我們採用GLR (Generalized Likelihood Ratio) 來發現異常點。在這上面我們也做了相應的改進，包括：
● Floating Threshold GLR, 基於新的輸入資料動態調整模型；
● 對於噪聲比較大的資料做去除異常點。

3.3.2異常檢測演算法2

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

在某個時刻t, 我們收到一個新的資料點，對於歷史每個資料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
報警觸發門檻通過Doob’s maximal inequality控制
Prob (∃ t :Mt>λ)<1λ
對於異常點，Martingale的值就會大於門檻值。

3.3.3異常檢測演算法3

這是一個簡單而非常有效的基於歷史資料的指數平滑演算法。
它首先基於歷史資料生成動態上下界：

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