StreamInsight 流處理體系結(jié)構(gòu)和熟悉的基于.NET 的開發(fā)平臺使開發(fā)人員可以快速實現(xiàn)強(qiáng)大而高效的事件處理應(yīng)用程序。本文我們主要就介紹一下利用Microsoft StreamInsight控制較大數(shù)據(jù)流的具體方法，接下來就讓我們一起來了解一下這部分內(nèi)容吧。

利用Microsoft StreamInsight 控制較大數(shù)據(jù)流

生產(chǎn)線的產(chǎn)量下降后，將容易出現(xiàn)用戶媒體流跳過這些流程，或者您的一個產(chǎn)品成為了“必需產(chǎn)品”的情況。真正的竅門是在這些情況發(fā)生時進(jìn)行識別，或根據(jù)以往趨勢對其做出預(yù)測。

成功預(yù)測這些情況需要使用近乎實時的方法。在對相關(guān)數(shù)據(jù)進(jìn)行提取、轉(zhuǎn)換并加載到 SQL Server Analysis Services (SSAS) 等傳統(tǒng)商業(yè)智能 (BI) 解決方案中時，情況早已發(fā)生改變。同樣，一些系統(tǒng)依靠請求-響應(yīng)模式來從事務(wù)性數(shù)據(jù)存儲（如 SQL Server Reporting Services 或 SSRS、報告）中請求已更新的數(shù)據(jù)，像這樣的系統(tǒng)總是在接近請求-輪詢間隔結(jié)束時運行陳舊數(shù)據(jù)。輪詢間隔通常是固定的，因此即使突然發(fā)生有趣的活動，消耗系統(tǒng)也不會知道，直到進(jìn)入下一個間隔。相反，消耗系統(tǒng)應(yīng)該在滿足趣味條件時連續(xù)收到通知。

在檢測新興趨勢時，時間間隔至關(guān)重要 - 在過去的五分鐘內(nèi)，一個特定項目發(fā)生了 100 次購買，顯而易見，這比過去五個月間的持續(xù)購買更能指示新興趨勢。SSAS 和 SSRS 等傳統(tǒng)系統(tǒng)需要開發(fā)人員通過事務(wù)性存儲中多維數(shù)據(jù)集或時間戳列中的單獨維度來自行跟蹤數(shù)據(jù)的及時性。理論上，用于識別新興情況的工具可能具有時間內(nèi)置的概念，并能提供使用該工具所需的豐富 API。

最后，對未來的準(zhǔn)確指示來源于對過去的分析。實際上，這就是傳統(tǒng) BI 的所有功能 - 對大量的歷史數(shù)據(jù)進(jìn)行匯總和分析，從而識別趨勢。遺憾的是，與更多的事務(wù)性系統(tǒng)相比，在使用這些系統(tǒng)時需要不同的工具和查詢語言。成功識別新興情況需要實現(xiàn)過去數(shù)據(jù)和當(dāng)前數(shù)據(jù)的無縫關(guān)聯(lián)。只有當(dāng)對這兩種數(shù)據(jù)使用相同的工具和查詢語言時，才可能實現(xiàn)這種緊密集成。

對于生產(chǎn)線監(jiān)視等特定情況，可通過存在的針對性極強(qiáng)的自定義工具來執(zhí)行這些功能，但是這些工具通常比較昂貴且用途并不廣泛。

為了防止生產(chǎn)線產(chǎn)量下降或確保您的產(chǎn)品定價合適，關(guān)鍵在于具有足夠的響應(yīng)能力，能夠根據(jù)情況的更改而進(jìn)行識別與調(diào)整。若要輕松快速地識別這些情況，歷史查詢和實時查詢應(yīng)使用相同的開發(fā)人員友好的工具集和查詢語言，系統(tǒng)應(yīng)該以近乎實時的方式來處理大量的數(shù)據(jù)（大約為每秒成百上千個事件），同時引擎應(yīng)該足夠靈活，能夠處理跨越多個問題域的情況。

幸運的是，存在這樣的工具。它稱為 Microsoft StreamInsight。

StreamInsight 體系結(jié)構(gòu)概述

StreamInsight 是一種復(fù)雜事件處理引擎，它每秒能夠處理成百上千的事件，且延遲極低。它可以由任何進(jìn)程（如 Windows 服務(wù)）托管，也可以直接嵌入任何應(yīng)用程序。StreamInsight 具有簡單的適配器模型，用于輸入和輸出數(shù)據(jù)，并且實時數(shù)據(jù)和歷史數(shù)據(jù)的查詢像任何其他來自任何 Microsoft .NET Framework 語言的程序集一樣使用獲取的相同 LINQ 語法。其作為 SQL Server 2008 R2 的一部分授予許可。

StreamInsight 的高級體系結(jié)構(gòu)非常簡單：通過輸入適配器從各種源收集事件。這些事件均通過查詢進(jìn)行分析和轉(zhuǎn)換，并且查詢結(jié)果通過輸出適配器分發(fā)給其他系統(tǒng)和人?！　D 1 顯示了這一簡單結(jié)構(gòu)。

圖 1 Microsoft StreamInsight 高級體系結(jié)構(gòu)

就像面向服務(wù)的體系結(jié)構(gòu)關(guān)注消息，而數(shù)據(jù)庫系統(tǒng)關(guān)注行一樣，StreamInsight 等復(fù)雜事件處理系統(tǒng)按照事件進(jìn)行組織。事件是簡單的數(shù)據(jù)段以及與該數(shù)據(jù)相關(guān)的時間 - 與一天中特定時間的傳感器讀數(shù)或股票行情價格相似。事件所攜帶的數(shù)據(jù)稱為它的負(fù)載。

StreamInsight 支持三種類型的事件。點事件是即時且不持續(xù)的事件。間隔事件是其負(fù)載與特定時間段相關(guān)的事件。邊緣事件與間隔事件相似，但當(dāng)邊緣事件到達(dá)時，其持續(xù)時間未知。而系統(tǒng)設(shè)置了開始時間，且事件實際上具有無限持續(xù)時間，直到另一個邊緣事件到達(dá)才會為這一事件設(shè)置結(jié)束時間。例如，速度計讀數(shù)可能為點事件，因為它不斷更改，但是超市的牛奶價格可能為邊緣事件，因為其關(guān)聯(lián)時間較長。當(dāng)牛奶的零售價格更改時（比如，由于分銷商定價發(fā)生更改），新價格的持續(xù)時間未知，因此，與間隔事件相比，邊緣事件要更為合適。稍后，當(dāng)分銷商再次更新其定價時，新的邊緣事件將覆蓋先前定價更改的持續(xù)時間，而另一個邊緣事件將設(shè)置新的價格以便繼續(xù)。

StreamInsight 中的輸入適配器和輸出適配器是適配器設(shè)計模式的抽象示例。StreamInsight 引擎在其自有的事件表示上運行，但是這些事件的實際來源可能有較大差異，范圍從專有接口到硬件傳感器到由企業(yè)的應(yīng)用程序生成的狀態(tài)消息。輸入適配器將源事件轉(zhuǎn)換為引擎能夠理解的事件流。

來自 StreamInsight 查詢的結(jié)果表示特定商業(yè)知識，且能夠高度專業(yè)化。將這些結(jié)果路由至最合適的地點，這點至關(guān)重要。輸出適配器可用于將事件的內(nèi)部表示轉(zhuǎn)換為打印到控制臺的文本、通過 Windows Communication Foundation (WCF) 發(fā)送到另一個系統(tǒng)以供處理的消息，甚至 Windows Presentation Foundation 應(yīng)用程序中圖表上的點。有關(guān)使用文本文件、WCF 和 SQL 等的示例適配器可從 streaminsight.codeplex.com 獲得。

StreamInsight Queries by Example

乍一看，StreamInsight 查詢似乎與從數(shù)據(jù)庫中查詢行相似，但是兩者之間存在重大差異。查詢數(shù)據(jù)庫時，系統(tǒng)會構(gòu)造并執(zhí)行查詢，同時返回結(jié)果。如果基礎(chǔ)數(shù)據(jù)發(fā)生更改，輸出并不會因為已運行查詢而受影響。數(shù)據(jù)庫查詢結(jié)果表示某一時刻的快照，可以通過請求-響應(yīng)模式使用。

StreamInsight 查詢?yōu)楝F(xiàn)有查詢。隨著新輸入事件的到達(dá)，查詢不斷響應(yīng)，并且根據(jù)需要創(chuàng)建新的輸出事件。

本文中的查詢示例來自可供下載的示例解決方案。這些示例開始較簡單，但隨著查詢語言新功能的引入，功能變得更加強(qiáng)大。所有查詢都使用同一負(fù)載類。以下是一個簡單類的定義，該類具有 Region 屬性和 Value 屬性：

 public class EventPayload  
 {   
 　　public string Region { get; set; }  
　　public double Value { get; set; }  
　   public override string ToString()   
 　　{   
 　　　　return string.Format("{0}\t{1:F4}", Region, Value);  
　　}  
 } 

示例應(yīng)用程序中的查詢使用一臺輸入適配器和一臺輸出適配器來進(jìn)行，輸入適配器可隨機(jī)生成數(shù)據(jù)，輸出適配器只需將各事件寫入控制臺。為清晰起見，對示例應(yīng)用程序中的適配器進(jìn)行了簡化。

若要運行每個查詢，請在示例解決方案中取消注釋 Program.cs 文件中的行，該示例解決方案可將查詢分配給稱為“template”的本地變量。

以下是一個基本查詢，它通過 Value 屬性來篩選事件：

var filtered =  
　　from i in inputStream  
　　where i.Value > 0.5  
　　select i; 

具有使用 LINQ 經(jīng)驗的任何開發(fā)人員應(yīng)該非常熟悉此查詢。因為 StreamInsight 使用 LINQ 作為它的查詢語言，因此此查詢與 LINQ to SQL 查詢類似，訪問數(shù)據(jù)庫或?qū)?IList 進(jìn)行內(nèi)存中篩選。當(dāng)事件從輸入適配器到達(dá)時，其負(fù)載將受到檢查，并且如果 Value 屬性的值大于 0.5，事件將被傳遞到輸出適配器，并在此將其打印到控制臺。

應(yīng)用程序運行時，可以看到事件不斷到達(dá)輸出中。這實際上是一個推模型。當(dāng)事件到達(dá)時，StreamInsight 會計算來自輸入的新輸出事件，這與數(shù)據(jù)庫等拉模型不同，在拉模型中，應(yīng)用程序必須定期輪詢數(shù)據(jù)源，以查看新數(shù)據(jù)是否已經(jīng)到達(dá)。這能與 Microsoft .NET Framework 4 中可用的 IObservable 支持完美結(jié)合，我們將在后續(xù)章節(jié)中對此進(jìn)行介紹。

使用推模型代替輪詢來處理連續(xù)數(shù)據(jù)是個非常好的主意，但是 StreamInsight 的真正功能體現(xiàn)在查詢時間相關(guān)的屬性上。當(dāng)事件通過輸入適配器到達(dá)時，它們獲得了一個時間戳。該時間戳可能來自數(shù)據(jù)源本身（假設(shè)事件表示歷史數(shù)據(jù)，且?guī)в杏糜诖鎯r間的顯示列），或者可以設(shè)置為事件到達(dá)的時間。實際上，時間是 StreamInsight 查詢語言中的第一個類。

查詢通常與標(biāo)準(zhǔn)數(shù)據(jù)庫查詢類似，標(biāo)準(zhǔn)數(shù)據(jù)庫查詢在尾部粘貼有時間限制符，如“每五秒”或“五秒的時間跨度上每三秒”。例如，以下是一個簡單查詢，它每五秒查詢一次 Value 屬性的平均值：

var aggregated =  
 　　from i in inputStream.TumblingWindow(TimeSpan.FromSeconds(5),   
 　　HoppingWindowOutputPolicy.ClipToWindowEnd)  
 　　select new { Avg = i.Avg(p => p.Value)}; 

數(shù)據(jù)窗口

因為時間概念是復(fù)雜事件處理系統(tǒng)的基礎(chǔ)必需概念，因此應(yīng)以簡單的方式來使用系統(tǒng)中查詢邏輯的時間組件，這點非常重要。StreamInsight 使用窗口概念來表示按時間分組。之前的查詢使用翻轉(zhuǎn)窗口。應(yīng)用程序運行時，查詢將每五秒生成單個輸出事件（窗口的大?。?。輸出事件表示前五秒的平均值。像 LINQ to SQL 或 LINQ to Object 一樣，聚合方法（如 Sum 和 Average）能夠?qū)磿r間分組的事件匯總為單個值，或可以使用 Select 將輸出投影成不同格式。

翻轉(zhuǎn)窗口只是另一種窗口類型的特例：跳躍窗口。跳躍窗口也有大小，但是它們也具有不等于其窗口大小的跳躍大小。這表示跳躍窗口可以互相重疊。

例如，窗口大小為五秒、跳躍大小為三秒的跳躍窗口將每三秒生成輸出（跳躍大?。?，提供前五秒的平均值（窗口大小）。它一次向前跳躍三秒，且持續(xù)五秒。圖 2 顯示分組為翻轉(zhuǎn)窗口和跳躍窗口的事件流。

圖 2 翻轉(zhuǎn)窗口和跳躍窗口

請注意，翻轉(zhuǎn)窗口并不重疊，但是對于跳躍窗口，如果跳躍大小小于窗口大小，則可以重疊。如果窗口重疊，事件將可能在多個窗口中結(jié)束，如同時存在于窗口 1 和窗口 2 中的第三個事件。邊緣事件（具有持續(xù)時間）也可能在窗口邊緣重疊，并在多個窗口中結(jié)束，如翻轉(zhuǎn)窗口中的倒數(shù)第二個事件。

另一種常見窗口類型為計數(shù)窗口。計數(shù)窗口包含特定數(shù)量的事件，而不是某一時間點或時間段內(nèi)的事件。要查詢最后三個到達(dá)的事件的平均數(shù)，可能需要使用計數(shù)窗口。計數(shù)窗口當(dāng)前的一個限制是不支持 Sum 和 Average 等內(nèi)置聚合方法。您必須創(chuàng)建用戶定義的聚合。下文會對這一簡單流程進(jìn)行介紹。

最后一種窗口類型為快照窗口。在邊緣事件的環(huán)境下，快照窗口最容易理解。每次事件的開始或結(jié)束即表示當(dāng)前窗口的完成和新窗口的開始。圖 3 顯示如何將邊緣事件分組為快照窗口。請注意每個事件邊界觸發(fā)窗口邊界的方式。E1 開始，w1 也開始。當(dāng) E2 開始時，w1 完成，而 w2 開始。下個邊緣是 E1 結(jié)束，使得 w2 完成，而 w3 開始。結(jié)果為三個窗口：包含 E1 的 w1，包含 E1 和 E2 的 w2 以及包含 E3 的 w3。事件分組為窗口后，它們會受到拉伸，從而使事件的開始與結(jié)束時間與窗口的相同。

圖 3 快照窗口

更多復(fù)雜查詢

在提供可用窗口與基本查詢方法（如地點、分組依據(jù)和排序依據(jù)）的情況下，可以進(jìn)行多種查詢。以下是一個查詢，其將輸入事件按地區(qū)分組，然后使用跳躍窗口來輸出最后一分鐘各個 Region 的負(fù)載 Value 的總和：

var payloadByRegion =   
　　 from i in inputStream   
　　 group i by i.Region into byRegion   
from c in byRegion.HoppingWindow(   
　 TimeSpan.FromMinutes(1),  
　 TimeSpan.FromSeconds(2),   
　　 HoppingWindowOutputPolicy.ClipToWindowEnd)  
　　 select new {  
　　Region = byRegion.Key,   
　　Sum = c.Sum(p => p.Value)}; 

這些窗口使用兩秒的跳躍大小，因此引擎每兩秒發(fā)送輸出事件。

因為查詢運算符是在 IQueryable 接口中定義的，因此可以撰寫查詢。以下代碼使用上一個查詢，其按地區(qū)查找總和，并計算總和最高的地區(qū)?？煺沾翱谠试S事件流按總和分類，因此可以使用 Take 方法獲取總和最高的地區(qū)：

var highestRegion =   
 // Uses groupBy query   
 　　(from i in payloadByRegion.SnapshotWindow(  
 　　SnapshotWindowOutputPolicy.Clip)  
 　　 from sumByRegion in i  
 　　orderby sumByRegion.Sum descending  
 　　select sumByRegion).Take(1); 

一般情況是有關(guān)快速移動事件（如傳感器中的讀數(shù)）到慢速移動或靜態(tài)參考數(shù)據(jù)（如傳感器的固定位置）流的查詢。查詢使用聯(lián)接來實現(xiàn)此目的。

StreamInsight 聯(lián)接語法與任何其他 LINQ 聯(lián)接相同，但有一點需要注意：當(dāng)事件的持續(xù)時間重疊時，它們才會聯(lián)接在一起。如果傳感器 1 在時間 t1 報告了一個值，但是有關(guān)傳感器 1 位置的參考數(shù)據(jù)僅對時間 t2 到 t3 有效，那么聯(lián)接將不匹配。持續(xù)時間的聯(lián)接條件并沒有明確寫入查詢定義中；這是 StreamInsight 引擎的基本屬性。使用靜態(tài)數(shù)據(jù)時，通常情況下，輸入適配器實際上將數(shù)據(jù)處理為帶有無限持續(xù)時間的邊緣事件。這樣將能成功完成到快速移動事件流的所有聯(lián)接。

通過聯(lián)接來關(guān)聯(lián)多個事件流是一個非常強(qiáng)大的概念。裝配線、石油生產(chǎn)設(shè)施或高容量網(wǎng)站通常不會因為隔離的事件而發(fā)生故障。一個用于觸發(fā)溫度警報的設(shè)備部件通常不會導(dǎo)致生產(chǎn)線癱瘓；生產(chǎn)線癱瘓可能由于多個原因造成，如溫度在某一持續(xù)時間段內(nèi)過高，同時某一工具使用過多，而操作員正在換班。

如果沒有聯(lián)接，隔離事件將不會有這么多的商業(yè)價值。通過對歷史數(shù)據(jù)使用聯(lián)接和 StreamInsight 查詢，用戶可以將隔離流與非常具體的監(jiān)控條件相關(guān)聯(lián)，然后進(jìn)行實時監(jiān)控。現(xiàn)有查詢能夠查找可能導(dǎo)致故障的情況，并自動生成可路由至系統(tǒng)的輸出事件，該系統(tǒng)知道如何使過熱的設(shè)備部件脫機(jī)，而不是等到該部件造成整條生產(chǎn)線停產(chǎn)。

在零售情況中，有關(guān)某段時間按項目劃分的銷售量的事件可以輸入到定價系統(tǒng)和客戶訂單歷史記錄中，從而確保每個項目具有最佳的定價，或決定在用戶結(jié)賬前向其推薦的項目。由于查詢易于創(chuàng)建、修改和撰寫，因此您可以從簡單的情況開始，并隨時間的流逝進(jìn)行優(yōu)化，從而增加業(yè)務(wù)價值。

用戶定義的聚合

StreamInsight 附帶最常見的聚合函數(shù)，包括 Count、Sum 和 Average。當(dāng)這些函數(shù)不夠時（或您需要在前文提到的計數(shù)窗口進(jìn)行聚合），StreamInsight 支持用戶定義的聚合函數(shù)。

要創(chuàng)建用戶定義的聚合，其流程包括兩個步驟：編寫實際聚合方法，然后通過擴(kuò)展方法將該方法公布到 LINQ。

進(jìn)行第一步時，如果聚合與時間無關(guān)，則從 CepAggregate<TInput, TOutput> 繼承，如果聚合與時間有關(guān)，則從 CepTimeSensitiveAggregate<TInput,TOutput> 繼承。這些抽象類具有單獨的實現(xiàn)方法，稱為 GenerateOutput。

 public class EveryOtherSum :  
 　　CepAggregate<double, double> {  
 　　　　public override double GenerateOutput(  
 　　　　　　IEnumerable<double> payloads) {  
 　　　　　　　　var sum = default(double);  
 　　　　　　　　var include = true;  
 　　　　　　　　foreach (var d in payloads) {  
 　　　　　　　　　　if (include) sum += d;  
　　　　　　　　　　include = !include;  
 　　　　　　　　}  
 　　　　　　　　return sum;  
 　　　　　　}  
 　　　} 

進(jìn)行第二步時，需要在 CepWindow<TPayload> 上創(chuàng)建擴(kuò)展方法，以便可以在查詢中使用您的聚合。CepUserDefinedAggregateAttribute 適用于擴(kuò)展方法，以便通知 StreamInsight 在哪里可以找到聚合的實現(xiàn)（在這種情況下，類是在第一步中創(chuàng)建的）。在可下載的示例應(yīng)用程序中，本流程兩個步驟的代碼均可在 EveryOtherSum.cs 文件中找到。

關(guān)于利用Microsoft StreamInsight控制較大數(shù)據(jù)流的相關(guān)知識就介紹到這里了，希望本次的介紹能夠?qū)δ兴斋@！

【編輯推薦】

1. 初學(xué)SQL Server數(shù)據(jù)庫的一些常用操作總結(jié)
2. SQL Server數(shù)據(jù)庫創(chuàng)建數(shù)據(jù)倉庫已分區(qū)表詳解
3. SQL Server與Access數(shù)據(jù)庫ASP代碼的比較詳解
4. SQL Server數(shù)據(jù)庫中bit字段類型使用時的注意事項
5. SQL Server數(shù)據(jù)庫timestamp數(shù)據(jù)類型相關(guān)知識介紹