李幸斌，程良倫

(廣東工業(yè)大學(xué)計(jì)算機(jī)學(xué)院，廣東廣州510006)

0 引言

制造業(yè)物聯(lián)網(wǎng)中，提高流數(shù)據(jù)處理的吞吐率具有重要意義。對(duì)流處理的研究已經(jīng)有很多，最開始的流處理方式是流數(shù)據(jù)庫［1］，它們以數(shù)據(jù)存儲(chǔ)為中心，提供豐富的查詢語言，然而這種事件處理的方式，由于要先將數(shù)據(jù)存入數(shù)據(jù)庫再取出，效率非常低。比較典型的流式數(shù)據(jù)庫有:以開源數(shù)據(jù)庫PostgreSQL［2］為基礎(chǔ)，通過體系結(jié)構(gòu)改造以支持連續(xù)查詢的TelegraphCQ［3］以及提供歷史查詢功能的Truviso［4］。之后出現(xiàn)了一些流處理系統(tǒng)，經(jīng)典的有 STREAM［5］，Aurora以及 PIPES［6］等。它們的共同特點(diǎn)是都采用了發(fā)布/訂閱模式，優(yōu)點(diǎn)是提高了處理速度，主要的缺點(diǎn)是查詢語言的表現(xiàn)力有限，只能執(zhí)行簡(jiǎn)單的選擇操作。數(shù)據(jù)流處理系統(tǒng)通常是分布式和高度并行的，盡管數(shù)據(jù)流處理系統(tǒng)查詢的效率很高，但它們不能處理多個(gè)事件模式的查詢，因此，不適合處理制造業(yè)物聯(lián)網(wǎng)中復(fù)雜多源的原始數(shù)據(jù)。

最近提出了一些基于主動(dòng)數(shù)據(jù)庫技術(shù)的新的流處理系統(tǒng)，被稱之為復(fù)雜事件處理(以下簡(jiǎn)稱CEP)。現(xiàn)有CEP項(xiàng)目有 SASE，SASE+［7］和 Cayuga［8］等。SASE 系統(tǒng)采用了基于本地序列操作符和管道查詢的數(shù)據(jù)流模型，使用關(guān)系運(yùn)算符來定義隨后到來的序列。在SASE中，一個(gè)查詢被轉(zhuǎn)化為一個(gè)非確定有窮自動(dòng)機(jī)，非確定有窮自動(dòng)機(jī)的每一個(gè)狀態(tài)有一個(gè)相應(yīng)的活動(dòng)實(shí)例棧保存匹配的事件。當(dāng)一個(gè)事件到達(dá)時(shí)，如果它匹配轉(zhuǎn)換條件，就將它壓入到合適的活動(dòng)實(shí)例棧中。如果事件到達(dá)接受狀態(tài)，活動(dòng)實(shí)例棧中的事件都被退棧了，并且模式匹配構(gòu)造被執(zhí)行完成，則模式匹配完成。與許多其他系統(tǒng)不同，SASE不僅會(huì)報(bào)告用戶感興趣的查詢結(jié)果，而且會(huì)報(bào)告匹配此查詢的所有事件，這在很大程度上增加了查詢處理的復(fù)雜度。SASE的主要局限性在于不能處理層狀結(jié)構(gòu)的復(fù)雜事件類型，即一個(gè)查詢的結(jié)果不能用作另一個(gè)查詢的輸入。康奈爾開發(fā)的Cayuga彌補(bǔ)了這個(gè)不足。Cayuga使用了傳統(tǒng)的發(fā)布/訂閱技術(shù)，因此，它支持大量并發(fā)的訂閱事件。Cayuga引擎使用單線程讀取數(shù)據(jù)和利用自動(dòng)機(jī)處理數(shù)據(jù)。自動(dòng)機(jī)允許對(duì)輸入數(shù)據(jù)進(jìn)行存儲(chǔ)，這使得新的輸入可以與先前存儲(chǔ)的事件做比較。Cayuga還使用了類似流水線的處理模型，該模型需要每個(gè)查詢結(jié)果實(shí)時(shí)輸出給下個(gè)處理過程使用。另外，Cayuga中采用了查詢優(yōu)化技術(shù)，將多個(gè)擁有相同時(shí)間戳的具有等價(jià)狀態(tài)的事件一起處理。然而，由于它的內(nèi)核是單線程的，這些優(yōu)化技術(shù)并沒有顯著提高Cayuga的性能，難以滿足制造物聯(lián)網(wǎng)中對(duì)事件流的實(shí)時(shí)響應(yīng)需求。

本文提出聚集活動(dòng)實(shí)例棧中的連接，并批量執(zhí)行序列構(gòu)造的方法來提高CEP查詢的吞吐率。仿真實(shí)驗(yàn)結(jié)果表明:該方法有效的提高了復(fù)雜事件處理的吞吐率。

1 SASE

首先解釋一下SASE中CEP的處理過程。本文假設(shè)窗口的大小是9，查詢模式是“SEQ(A，B，D)”，輸入事件序列如下

其中，第一個(gè)字母表示事件類型，第二個(gè)字符表示到達(dá)的時(shí)間。例如:“c2”表示事件類型為“C”，時(shí)間戳為“2”。“SEQ(A，B，D)”表明B在A的后面。本文遵循“跳過直到匹配”的策略［1，2］，因此，A 和B 之間的事件、B 和 D 之間的事件會(huì)被忽略。所以，從上面事件中獲取的結(jié)果是(a1，b3，d5)。

下面描述SASE中輸入事件如何被處理到接收狀態(tài):

1)查詢轉(zhuǎn)化為NFA。

2)每一個(gè)NFA都會(huì)有一個(gè)相應(yīng)的AIS，AISs存儲(chǔ)事件匹配的狀態(tài)，當(dāng)一個(gè)事件輸入到NFA，如果它轉(zhuǎn)換到其他狀態(tài)而不是當(dāng)前狀態(tài)，則將這個(gè)事件將被壓入到AIS。

3)當(dāng)NFA到達(dá)接受狀態(tài)時(shí)，它使用連接構(gòu)造模式匹配序列。

4)在完成模式匹配序列構(gòu)造之后，調(diào)用模式匹配序列構(gòu)造的事件將從接受狀態(tài)的AIS中刪除。之后將重復(fù)步驟(2)～ (4)。

根據(jù)上面的描述，可以簡(jiǎn)明地描述SASE中NFA的行為:它首先接收一個(gè)事件，然后，如果事件引起狀態(tài)轉(zhuǎn)換，NFA壓入事件到一個(gè)合適的AIS，然后創(chuàng)建一個(gè)從當(dāng)前事件到之前AIS中事件的連接。如果NFA到了接受狀態(tài)，模式匹配構(gòu)造器被觸發(fā)，將生成匹配序列。用算法1描述過程如下:

算法1

1:Wait for an event;

2:Receive an event e;

3:IF(e does not invoke state transition)

4:Go to line 1;

5:ENDIF

6:Invoke state transition and push e onto appropriate AIS;

7:Create a link from e to an event in the previous stack;

8:IF(current state is acceptance state)

9:Construct pattern occurrences using links originated from e;

10:Delete e and its link;

11:ENDIF

12:Drop events in the outside of window from all AISs;

13:Go to line 1.

2 建議的方法

觀察算法1，將發(fā)現(xiàn)算法的第8～11行有一個(gè)潛在的瓶頸:模式匹配序列構(gòu)造處理需要花費(fèi)長(zhǎng)時(shí)間。因此，如果減少模式匹配序列的構(gòu)造花費(fèi)，就可以獲得高的吞吐率。在本文中，提出了聚集AISs中的連接，批量執(zhí)行模式匹配序列構(gòu)造的方法。因?yàn)榫奂B接減少了檢索連接的花費(fèi)，它將加快查詢處理。下面，解釋這種方法如何進(jìn)行“SEQ(A，B，D)”查詢。窗口的大小同樣設(shè)為9，輸入事件如下

下面一步一步描述本文提出的方法:

1)一個(gè)事件輸入NFA，如果它引起狀態(tài)轉(zhuǎn)換，將事件壓入下一狀態(tài)的AIS，然后創(chuàng)建一個(gè)連接到這個(gè)事件。連接的目的事件的時(shí)間戳必須小于源事件的時(shí)間戳，目的事件必須是滿足條件的事件中時(shí)間戳最大的。

2)有相同連接(RIP)的事件被聚集并打包成一個(gè)簇:在這個(gè)例子中，“d7”和“d9”有相同的連接，因此，這兩個(gè)連接打包到一個(gè)簇。

3)從連接中檢測(cè)是否發(fā)生模式匹配:在本例中，d7連接的目的事件是b3和b6.d9連接的目的事件與d7相同。因此，模式匹配構(gòu)造只被執(zhí)行1次，在SASE中將被執(zhí)行2次。在NFA達(dá)到接受狀態(tài)時(shí)，本文的方法并沒有調(diào)用模式匹配序列構(gòu)造器，而是批量執(zhí)行模式匹配序列構(gòu)造。在本例中，將構(gòu)造以下模式匹配序列:〈a1，b3，d5〉，〈a1，b3，d7〉，〈a1，b6，d7〉，〈a4，b6，d7〉，〈a1，b3，d9〉，〈a1，b6，d9〉，和〈a4，b6，d9〉。

4)刪除接收狀態(tài)的AISs中的事件:“d9”之后生成的事件，不滿足目的事件的時(shí)間戳需大于源事件的時(shí)間戳的連接條件。因此，d5，d7，d9是非必需事件，刪除它們。

5)刪除過期事件:窗口之外的事件將過期，因?yàn)榇翱诘拇笮∈?，例如:最后的事件是“d9”，當(dāng)下一個(gè)事件到來時(shí)，“a1”就過期了，需要?jiǎng)h除“a1”。之后，回到步驟(1)繼續(xù)執(zhí)行。用算法2描述過程如下:

算法2

1:Wait for an event;

2:Receive an event e;

3:IF(current time step%window size is 0)

4:Construct pattern occurrences using links originated from events in the AISfor acceptance state;

5:Delete expired events from all the AISs;

6:Go to line 1;

7:ENDIF

8:IF(e does not invoke state transition)

9:Go to line 1;

10:ENDIF

11:Invoke state transition and push e onto appropriate AIS;

12:IF(link destination of e already exists for e')

13:Merge e and e'and pack them as a cluster;

14:ELSE

15:Create a link from e to an event in the previous stack;

16:ENDIF

17:Go to line 1.

與SASE不同的是，本文的方法在接收狀態(tài)的AIS上壓入事件時(shí)，可能不執(zhí)行模式匹配構(gòu)造，而是批量構(gòu)建模式匹配序列，如算法2中3～7行所述。為了減少批量構(gòu)造的代價(jià)，聚集連接成簇，如算法2中12～14行所述。

3 性能分析

為了驗(yàn)證本文方法的有效性，下面比較了本文的方法和SASE中的傳統(tǒng)方法的吞吐率(計(jì)算兩種方法處理多個(gè)事件的時(shí)間，然后計(jì)算每秒的吞吐率)。實(shí)驗(yàn)操作系統(tǒng)為Windows XP Professional;內(nèi)存為3GB;CPU為Intel Core2Duo E8400;編程語言為Java(JRE 1.7.0_04)。

在實(shí)驗(yàn)中，事件數(shù)設(shè)置為10 000，窗口大小從500～1000進(jìn)行改變，使用四種類型的事件:A，B，C，D。通過隨機(jī)數(shù)生成器對(duì)4取模實(shí)現(xiàn)四種類型事件發(fā)生的可能性都是25%。使用的查詢語句為:“SEQ(A，B，D)”和“SEQ(A，B，D，C)”。

圖1是查詢“SEQ(A，B，D)”的實(shí)驗(yàn)結(jié)果。由圖可見所提出的方法的吞吐率高于傳統(tǒng)方法。最小的性能提高是窗口大小為500時(shí)，吞吐率為傳統(tǒng)方法的1.097倍;另一方面，當(dāng)窗口大小為1000時(shí)，獲得最大的吞吐率，此時(shí)是傳統(tǒng)方法的1.551倍。

圖1 SEQ(A，B，D)的結(jié)果Fig 1 Result of SEQ(A，B，D)

隨著窗口大小的增大，本文提出的方法和傳統(tǒng)的方法的吞吐率都減少了。這是因?yàn)殡S著窗口大小的增大，給了足夠的時(shí)間使模式匹配發(fā)生，從而使模式匹配發(fā)生的數(shù)量增加。因此，長(zhǎng)的窗口趨向于生成更多的模式匹配序列，顯然，這需要更多的資源，從而使處理新事件的模塊的執(zhí)行機(jī)會(huì)減少了，吞吐率就下降了。

圖2是查詢“SEQ(A，B，D，C)”的實(shí)驗(yàn)結(jié)果。它的吞吐率低于查詢“SEQ(A，B，D)”。最小的性能提高是在窗口大小為1000時(shí)，提高了1.55倍。在窗口大小為700時(shí)，獲得最大的性能提升，達(dá)到了1.62倍。

圖2 SEQ(A，B，D，C)的結(jié)果Fig 2 Result of SEQ(A，B，D，C)

與查詢“SEQ(A，B，D)”相比，查詢“SEQ(A，B，D，C)”吞吐率低的原因?yàn)?查詢“SEQ(A，B，D，C)”的 AISs中的事件數(shù)量更多，遍歷連接的時(shí)間更長(zhǎng)，模式匹配序列構(gòu)造的花銷更大。查詢“SEQ(A，B，D，C)”對(duì)應(yīng)的 NFA的結(jié)點(diǎn)數(shù)為4，這比之前的查詢高出了25%，導(dǎo)致了性能的巨大下降。

模式匹配發(fā)生數(shù)隨窗口大小的變化如圖3，可見模式匹配發(fā)生數(shù)正比于模式的長(zhǎng)度和窗口的大小。因此，吞吐率反比于窗口大小和模式的長(zhǎng)度。

圖3 吞吐率隨窗口的變化Fig 3 Change of output with window size

采用查詢“SEQ(A，B，D)”衡量聚集連接的有效性。輸入下面的事件作為一個(gè)分簇?zé)o效的例子，它沒有生成簇

a1，b2，d3，b4，d5，a6，b7，d8，b9，d10，….另外，輸入一個(gè)分簇有效的例子

a1，b2，b3，b4，d5，d6，d7，d8，d9，d10，a11，….

所有“D”類型的事件都在窗口中被分簇。圖4展示了分簇?zé)o效的仿真結(jié)果。圖5展現(xiàn)了分簇有效的仿真結(jié)果，在最好的時(shí)候達(dá)到了5.24倍(此時(shí)的窗口大小為1000)的性能提升。

4 結(jié)論

本文提出了一種通過聚集連接來提高CEP查詢吞吐率的方法。在分簇有效的情況下，本文的方法相比于SASE，達(dá)到了5.24倍的性能提升，這證明了聚集連接對(duì)提高吞吐率的有效性。由此得出結(jié)論:本文提出的聚集連接的方法對(duì)提高CEP查詢的吞吐率是有效的。

圖4 分簇?zé)o效的仿真結(jié)果Fig 4 Simulation result of ineffective clustering

圖5 分簇有效的仿真結(jié)果Fig 5 Simulation result of effective clustering

［1］Wu E，Diao Y，Rizvi S.High-performance complex event processing over streams［C］∥Proceedings of the 2006 ACM SIGMOD International Conference on Management of Data，ACM，2006:407-418.

［2］Arasu A，Babu S，Widom J.The CQL continuous query language:Semantic foundations and query execution［J］.The VLDB Journal—The International Journal on Very Large Data Bases，2006，15(2):121-142.

［3］Chandrasekaran S，Cooper O，Deshpande A，et al.TelegraphCQ:Continuous dataflow processing［C］∥Proceedings of the 2003 ACM SIGMOD International Conference on Management of Data，ACM，2003:668-668.

［4］Chen J，DeWitt DJ，Tian F，et al.NiagaraCQ:A scalable continuous query system for Internet databases［C］∥ACM SIGMOD Record，ACM，2000:379-390.

［5］Abadi D J，Carney D，etintemel U，et al.Aurora:A new model and architecture for data stream management［J］.The VLDB Journal—The International Journal on Very Large Data Bases，2003，12(2):120-139.

［6］Motwani R，Widom J，Arasu A，et al.Query processing，resource management，and approximation in a data stream management system［C］∥CIDR 2003，Stanford Info Lab，2002.

［7］Diao Y，Immerman N，Gyllstrom D.SASE+:An agile language for kleene closure over event streams［J/OL］.［2012—12—23］.http:∥archive，systems，ethz.ch/www，dbis.ethz.ch/education/ws0708/adv_top_infsyst/papers/sase_tr07，pdf，2007.

［8］Demers A J，Gehrke J，Panda B，et al.Cayuga:A general purpose event monitoring system［C］∥International Conference on Innovation Database Research，Online Proceedings，2007:412-422.