基于S4框架的并行復(fù)雜事件處理系統(tǒng)

陳皓，李瑜，虎嵩林，梁英

（1. 中國(guó)科學(xué)院 計(jì)算技術(shù)研究所，北京 100190；2. 中國(guó)科學(xué)院 研究生院，北京 100049）

1 引言

當(dāng)今社會(huì)是一個(gè)信息爆炸的社會(huì)。由于離散數(shù)據(jù)源的增加，如標(biāo)簽、微博、傳感技術(shù)等的發(fā)展，使得信息處理系統(tǒng)所需要進(jìn)行處理的數(shù)據(jù)量大大增加。另外，社交網(wǎng)絡(luò)、高頻交易、實(shí)時(shí)搜索等新應(yīng)用的出現(xiàn)，對(duì)處理速度的要求達(dá)到了傳統(tǒng)處理系統(tǒng)的極限。由于海量數(shù)據(jù)是具有時(shí)效性的，最佳的解決方法不是先把數(shù)據(jù)在數(shù)據(jù)庫(kù)中緩存起來(lái)再一批一批地處理，而是每當(dāng)數(shù)據(jù)出現(xiàn)時(shí)便進(jìn)行一次處理，實(shí)時(shí)地處理數(shù)據(jù)。復(fù)雜事件處理技術(shù)[1,2](CEP, complex event processing)作為信息處理系統(tǒng)的新興技術(shù)，具有高吞吐量、短延遲和復(fù)雜計(jì)算的特點(diǎn)。它采用以事件驅(qū)動(dòng)為基礎(chǔ)的架構(gòu)，可處理不同事件源、不同類型的事件。它通過(guò)分析事件間的關(guān)系，利用過(guò)濾、關(guān)聯(lián)、聚合、排序等技術(shù)，將輸入簡(jiǎn)單事件同其他事件聯(lián)系起來(lái)進(jìn)行檢測(cè)，在特定的上下文中分析計(jì)算，最終由簡(jiǎn)單事件產(chǎn)生輸出復(fù)雜事件。

復(fù)雜事件處理系統(tǒng)所需處理的數(shù)據(jù)流量往往非常大，然而用戶卻希望實(shí)時(shí)地得到結(jié)果。面對(duì)海量而關(guān)系復(fù)雜的信息，系統(tǒng)還須進(jìn)行快速地計(jì)算繼而快速?zèng)Q策，這對(duì)系統(tǒng)的吞吐量提出了很高的要求，單一節(jié)點(diǎn)的處理速度往往難以滿足需要。

MapReduce[3,4]是目前應(yīng)用極為廣泛的并行軟件構(gòu)架。它使用鍵值對(duì)描述數(shù)據(jù)，將復(fù)雜的運(yùn)算分成“Map（映射）”和“Reduce（化簡(jiǎn)）”2個(gè)獨(dú)立步驟進(jìn)行處理。Map和Reduce都是對(duì)其輸入數(shù)據(jù)中的元素進(jìn)行獨(dú)立地操作，所以這2個(gè)運(yùn)算步驟都是可以并行的，但Reduce操作要等待所有Map操作完成才能開(kāi)始。MapReduce模型可以很容易地將多個(gè)通用批處理任務(wù)和操作在大規(guī)模集群上并行化。但MapReduce是批量處理數(shù)據(jù)的，它將輸入數(shù)據(jù)切成小的片段，每隔一個(gè)周期就啟動(dòng)一次MapReduce任務(wù)。處理任務(wù)不是常駐服務(wù)，數(shù)據(jù)也不是實(shí)時(shí)流入，任務(wù)的分割難以滿足數(shù)據(jù)流實(shí)時(shí)處理的需求。因此，MapReduce框架不是并行CEP的合理方案。

Storm[5]是一種分布式實(shí)時(shí)計(jì)算系統(tǒng)框架，它由一個(gè)主節(jié)點(diǎn)和多個(gè)工作節(jié)點(diǎn)組成。主節(jié)點(diǎn)用于分配代碼、布置任務(wù)和故障檢測(cè)。工作節(jié)點(diǎn)用于監(jiān)聽(tīng)工作、開(kāi)始并終止工作進(jìn)程。它以有向圖定義系統(tǒng)的邏輯拓?fù)浣Y(jié)構(gòu)，每個(gè)節(jié)點(diǎn)將以指定的進(jìn)程數(shù)動(dòng)態(tài)分布在集群中，而事件消息按照預(yù)定義的分組方式分發(fā)至邏輯節(jié)點(diǎn)對(duì)應(yīng)的處理進(jìn)程。作為編程框架，它實(shí)現(xiàn)了事件驅(qū)動(dòng)的實(shí)時(shí)處理，但未提供對(duì)復(fù)雜事件處理中事件關(guān)系處理的支持，需要用戶根據(jù)場(chǎng)景自行開(kāi)發(fā)，其并行方式也需要用戶按照經(jīng)驗(yàn)自行配置，普通用戶難以使用。

本文針對(duì)復(fù)雜事件處理對(duì)吞吐量的需求，將通用CEP系統(tǒng)搭建在使用S4[6]并行處理框架的集群上，使用基于操作符的負(fù)載分流方法，將事件流分散到多處理節(jié)點(diǎn)上并行處理。既方便了普通用戶對(duì)CEP系統(tǒng)的配置，又提高了CEP系統(tǒng)的處理效率。

2 并行CEP系統(tǒng)

復(fù)雜事件處理通常按照應(yīng)用場(chǎng)景的需求，由一系列的事件生產(chǎn)者、處理代理、事件消費(fèi)者組成。邏輯上是一個(gè)有向圖，事件按照邊進(jìn)行傳遞，每個(gè)代理表示對(duì)于事件的一個(gè)操作。事件從生產(chǎn)者處被生產(chǎn)出來(lái)，最終流到了消費(fèi)者，從而完成了處理過(guò)程。

把操作符看做對(duì)于復(fù)雜事件處理的基本功能單位。各種操作符組合，以共同完成對(duì)于復(fù)雜時(shí)間的處理。將事件處理中的處理邏輯抽象出來(lái)，組合成一個(gè)一個(gè)的事件代理，而不是按照具體應(yīng)用程序定制處理過(guò)程，這種方法更為實(shí)用和強(qiáng)大。設(shè)計(jì)了14種操作符，這些操作符可過(guò)濾事件，對(duì)事件進(jìn)行多種變形和提供流上的模式識(shí)別功能以及一些簡(jiǎn)單的系統(tǒng)輔助功能。不僅使單個(gè)操作符的功能具有實(shí)用性，而且使多個(gè)操作符組合后能夠滿足對(duì)于事件分析的需求。在操作符的基礎(chǔ)上進(jìn)行CEP場(chǎng)景建模，使普通業(yè)務(wù)人員也能高效地定義配置CEP系統(tǒng)。

S4（simple scalable streaming system）是一個(gè)去中心的、分布式的、可擴(kuò)展的流式處理系統(tǒng)。類似于Storm的消息分組策略，S4框架通過(guò)對(duì)流事件的關(guān)鍵字進(jìn)行散列計(jì)算，得到處理事件的節(jié)點(diǎn)編號(hào)，再將事件發(fā)送到對(duì)應(yīng)節(jié)點(diǎn)進(jìn)行處理。通過(guò)修改S4框架的事件分流機(jī)制，基于操作符的不同，使用不同的分流策略，實(shí)現(xiàn)了并行的通用CEP系統(tǒng)。圖1描繪了其系統(tǒng)結(jié)構(gòu)。圖中箭頭方向?yàn)槭录飨颉Ｌ幚砉?jié)點(diǎn)通過(guò)網(wǎng)絡(luò)連接在一起，單個(gè)處理節(jié)點(diǎn)由CEP引擎、事件接收器、事件發(fā)送器和事件分流器組成。

事件接收器用于接收其網(wǎng)絡(luò)層發(fā)來(lái)的或處理節(jié)點(diǎn)內(nèi)部的事件消息，它將按照事件流名稱將事件放入不同的流中，交給CEP引擎處理；CEP引擎以操作符為基本運(yùn)行單位，事件根據(jù)其流的類型交由特定的操作符處理，處理結(jié)果將發(fā)往分流器；分流器應(yīng)用負(fù)載分流策略，基于目標(biāo)操作符和事件內(nèi)容進(jìn)行運(yùn)算，決定該事件由哪個(gè)節(jié)點(diǎn)處理。如果是本地節(jié)點(diǎn)，事件將進(jìn)入本地的接收器，重新發(fā)送給CEP引擎；如果是其他節(jié)點(diǎn)，將送至事件發(fā)射器，通過(guò)網(wǎng)絡(luò)發(fā)送給其他節(jié)點(diǎn)；發(fā)射器負(fù)責(zé)將事件發(fā)送給集群上的其他節(jié)點(diǎn)。S4初始版本只支持UDP的發(fā)送方式，將其改為帶有隊(duì)列的TCP方式，雖然增加了集群內(nèi)部連接數(shù)，但防止了事件的丟失。

該系統(tǒng)通過(guò)分流器，將事件分流至不同的處理節(jié)點(diǎn)，使得事件流得以在集群中多機(jī)并行處理，提高了系統(tǒng)整體吞吐。在系統(tǒng)節(jié)點(diǎn)部署時(shí)，各個(gè)CEP引擎中使用相同的操作符配置。節(jié)點(diǎn)部署的物理機(jī)器可手工配置，也可通過(guò)Zookeeper[7]協(xié)同自動(dòng)分配。通過(guò)提供備用節(jié)點(diǎn)，還可提高系統(tǒng)的可靠性。

圖1 并行CEP結(jié)構(gòu)(4節(jié)點(diǎn))

3 基于操作符的負(fù)載分流優(yōu)化方法

當(dāng)面對(duì)高速數(shù)據(jù)流時(shí)，如果能將負(fù)載均勻地分流到集群的各個(gè)節(jié)點(diǎn)，并行地處理事件流，則可以增加系統(tǒng)的吞吐量。然而復(fù)雜事件處理的特殊性在于事件往往不是孤立地存在，在處理中需要根據(jù)事件流上下文或其他事件流進(jìn)行決策，單純地拆分?jǐn)?shù)據(jù)流到不同節(jié)點(diǎn)，會(huì)影響事件間的關(guān)系，導(dǎo)致處理出錯(cuò)。

基于操作符的負(fù)載分流方法，根據(jù)目標(biāo)操作符的不同，使用不同的分流策略，實(shí)現(xiàn)動(dòng)態(tài)分流。本節(jié)將給出6種操作符的分流方案。

操作符的輸入流和輸出流均是由事件組成的，將事件表示為(K,V)的集合。輸入流事件表示為(Ki,Vi)，輸出事件流表示為(Ko, Vo)，中間結(jié)果表示為(Km,Vm)。每個(gè)操作符都是在輸入流上進(jìn)行計(jì)算，這個(gè)計(jì)算過(guò)程可以用函數(shù)operator來(lái)表示，最終將產(chǎn)生的結(jié)果放入輸出流。操作符的這一處理過(guò)程用如下的表達(dá)式來(lái)表示

1) 過(guò)濾操作符。首先使用分流器輸入分散到各處理節(jié)點(diǎn)，然后每個(gè)處理節(jié)點(diǎn)分別進(jìn)行原本的過(guò)濾操作，最后合并結(jié)果。這一分流過(guò)程可表示為

2) 映射操作符。首先使用分流器輸入分散到各處理節(jié)點(diǎn)，然后每個(gè)處理節(jié)點(diǎn)分別進(jìn)行原本的映射操作，最后將結(jié)果合并。這一分流過(guò)程可表示為

3) 分裂操作符。首先使用分流器輸入分散到各處理節(jié)點(diǎn)，然后每個(gè)處理節(jié)點(diǎn)分別進(jìn)行原本的分裂操作，最后須按原分裂目標(biāo)合并結(jié)果，這一分流過(guò)程可表示為

4) 合并操作符。首先使用分流器輸入分散到各處理節(jié)點(diǎn)，然后每個(gè)處理節(jié)點(diǎn)分別進(jìn)行原本的合并操作，最后將結(jié)果合并，這一分流過(guò)程可表示為

5) 查詢操作符。查詢表和物化窗口都屬于輔助表操作符，它們?cè)趦?nèi)存中建立一塊數(shù)據(jù)的緩存，供一個(gè)或者多個(gè)查詢操作符進(jìn)行插入、查詢、更新或刪除操作。分流的方案是將插入表的輸入流數(shù)據(jù)通過(guò)分流器分散到各處理節(jié)點(diǎn)，在每個(gè)節(jié)點(diǎn)上建立表，其他操作的輸入流需要復(fù)制到每個(gè)節(jié)點(diǎn), 最后將結(jié)果合，并這一分流過(guò)程可表示為

6) 聚集操作符。它根據(jù)2個(gè)及2個(gè)以上輸入流的屬性值相等性進(jìn)行連接操作。使用分流器將輸入流上鍵值屬于同一個(gè)范圍的事件分流到相同節(jié)點(diǎn)上，各節(jié)點(diǎn)再分別進(jìn)行聚集操作最終將結(jié)果合并。這一分流過(guò)程可表示為

在實(shí)際應(yīng)用中，不需要將結(jié)果進(jìn)行和并，因?yàn)椴僮鞣沁B續(xù)進(jìn)行流水線作業(yè)的，結(jié)果直接通過(guò)分流器進(jìn)入下一個(gè)分流階段。對(duì)于無(wú)法分流的操作符，如依據(jù)事件到達(dá)順序進(jìn)行計(jì)算的操作符，仍可配置分流器分流到指定的節(jié)點(diǎn)。

基于操作符特點(diǎn)進(jìn)行動(dòng)態(tài)地分流，既使分流工作對(duì)用戶透明地自動(dòng)完成，也使分流工作有針對(duì)性，有助于減輕處理節(jié)點(diǎn)的負(fù)載，整體上提高系統(tǒng)效率。

4 實(shí)驗(yàn)結(jié)果與分析

為了驗(yàn)證并行CEP系統(tǒng)的性能，在8臺(tái)機(jī)器構(gòu)成的集群上進(jìn)行了實(shí)驗(yàn)，每臺(tái)機(jī)器的配置為主頻1GHz，8核處理器，8G內(nèi)存。

4.1 實(shí)驗(yàn)配置

實(shí)驗(yàn)主要測(cè)試的是事件處理吞吐量。它反應(yīng)了系統(tǒng)在單位時(shí)間內(nèi)能處理的事件的規(guī)模，用每秒鐘處理的事件數(shù)量來(lái)量化表示。計(jì)算的方法是用總共處理的事件數(shù)量除以處理的時(shí)間。

數(shù)據(jù)來(lái)自國(guó)家電網(wǎng)智能園區(qū)的用電數(shù)據(jù)來(lái)模擬輸入的事件流。事件輸入流包括電表電量流、電價(jià)流和企業(yè)信息流3個(gè)。其中最重要的是電表電量流，它是每10s采集一次的電表讀數(shù)。使用了200塊電表一天內(nèi)采集的所有讀數(shù)，共1 728 000條數(shù)據(jù)。實(shí)驗(yàn)的方法是在是持續(xù)不停地向并行CEP系統(tǒng)中發(fā)送電表數(shù)據(jù)，在輸出端進(jìn)行吞吐量的度量。

4.2 實(shí)驗(yàn)結(jié)果

實(shí)驗(yàn)實(shí)時(shí)計(jì)算指定時(shí)間間隔內(nèi)各個(gè)企業(yè)的電費(fèi)。采用了1～8臺(tái)機(jī)器對(duì)輸入流進(jìn)行相同邏輯的復(fù)雜事件處理。3個(gè)查詢所包括的操作符如圖2所示，所測(cè)得的系統(tǒng)的吞吐量如圖3所示。

圖2 查詢所包含的操作符數(shù)量

圖3 不同節(jié)點(diǎn)數(shù)的事件吞吐量

由圖2和圖3可見(jiàn)，查詢一、查詢二和查詢?nèi)诠?jié)點(diǎn)數(shù)目增加的情況下吞吐量有顯著的提升。并且，吞吐量當(dāng)?shù)竭_(dá)峰值后，由于節(jié)點(diǎn)的數(shù)目的增加會(huì)大大增加消息傳遞的通信開(kāi)銷，吞吐量會(huì)減少。由于吞吐量同查詢的復(fù)雜程度相關(guān)，可見(jiàn)查詢一的峰值吞吐量最大，而查詢?nèi)姆逯低掏铝孔钚　６捎谔幚砉?jié)點(diǎn)數(shù)目越多，節(jié)點(diǎn)間的通信開(kāi)銷所占比重越大。并且，受限于集群的網(wǎng)絡(luò)環(huán)境，通信開(kāi)銷極大地影響了系統(tǒng)的吞吐量。所以，當(dāng)節(jié)點(diǎn)的數(shù)目繼續(xù)增加時(shí)，查詢?nèi)耐掏铝繙p少得更為明顯。

對(duì)6種操作符的負(fù)載分流性能進(jìn)行了評(píng)測(cè)。對(duì)于每個(gè)操作符，分別在不同分流的數(shù)目情況下進(jìn)行了實(shí)驗(yàn)。以下實(shí)驗(yàn)結(jié)果均是在4個(gè)節(jié)點(diǎn)的分布式環(huán)境下得到的。

圖4顯示了不同的操作符所設(shè)計(jì)的分流方案所包含的操作符個(gè)數(shù)。圖5顯示了實(shí)驗(yàn)所測(cè)得的不同操作符在不同的分流方案下運(yùn)行所得的吞吐量。由圖4和圖5可知，在分流的數(shù)量增加的情況下，系統(tǒng)的吞吐量有所提升。而由于分流數(shù)量增加，事件的通信開(kāi)銷會(huì)增加，所以當(dāng)分流數(shù)量進(jìn)一步增加的情況下，系統(tǒng)的吞吐量會(huì)降低。由于所涉及的操作符的運(yùn)算開(kāi)銷相差不大，所以各個(gè)操作符的實(shí)驗(yàn)結(jié)果無(wú)明顯區(qū)別。

圖4 負(fù)載分流后所得的實(shí)際操作符數(shù)

圖5 操作符在不同的負(fù)載分流方案下的吞吐量

實(shí)驗(yàn)結(jié)果表明，并行CEP系統(tǒng)的吞吐量在一定范圍內(nèi)隨著集群處理節(jié)點(diǎn)數(shù)的增加線性增長(zhǎng)，數(shù)倍于單機(jī)的處理能力。

5 結(jié)束語(yǔ)

本文針對(duì)復(fù)雜事件處理系統(tǒng)要求處理海量事件、進(jìn)行復(fù)雜計(jì)算和簡(jiǎn)易直觀建模的需求，設(shè)計(jì)了14種操作符給出了6個(gè)操作符的負(fù)載分流方案，然后在S4框架下搭建了并行CEP系統(tǒng)的方法，實(shí)驗(yàn)證明并行CEP系統(tǒng)能有效地提高系統(tǒng)的吞吐量。進(jìn)一步工作將深入分析其他操作符的分流方案，提高系統(tǒng)的并行率。

[1] LUCKHAM D C, VERA J. An event-based architecture definition language[J]. IEEE Transactions on Software Engineering, 1995, 21(9):717-734.

[2] LUCKHAM D C. The Power of Events: An Introduction to Complex Event Processing in Distributed Enterprise Systems[M]. Boston: Addison-Wesley Professional, 2002.

[3] DEAN J, GHEMAWAT S. Mapreduce: simplified data processing on large clusters[J]. Commun ACM, 2008, 51(1): 107-113.

[4] CONDIE T, CONWAY N, ALVARO P, etal. Map reduce online[A].Proceedings of the 7th USENIX Conference on Networked Systems Design and Implementation[C]. Berkeley, San Jose, CA, USA, 2010. 21.

[5] Storm wiki[EB/OL].http://github.com/nathanmarz/storm/wiki,2012.

[6] NEUMEYER L, ROBBINS B, NAIR A, etal. S4: distributed stream computing platform[A]. The 10th IEEE International Conference on Data Mining Workshops[C]. Sydney, Australia, 2010. 170-177.

[7] Zookeeper[EB/OL]. http://zookeeper.apache.org/,2012.