SODBASE CEP學習（五）：流式計算中的類SQL語言EPL

阿新 • • 發佈：2019-01-15

（本文中類SQL語句建模、單元測試建議使用SODBASE Studio，參考示例見視訊教程。

SODBASE CEP中，類SQL語言EPL（事件處理語言）也叫做SODSQL。其基本寫法為

CREATE QUERY 查詢名稱
SELECT 查詢欄位
FROM  流
PATTERN 複雜事件模式
WHERE 條件
WITHIN 時間視窗大小

比傳統SQL就多了兩個東西PATTERN和WITHIN

PATTERN：複雜事件表示式，由事件類名和操作符構成，操作符含義參見下文。

CEP理論通常規定事件有兩個時間戳，即開始時間和結束時間。SODBASE CEP對應資料的_start_time_和_end_time_屬性。CEP理論對事件也作了區分：基本事件和複雜事件。基本事件往往代表著原始資料，開始時間等於結束時間。可以這樣理解，如果說流是一張無限長的關係表，那基本事件就是關係表中一條條記錄，並帶了時間戳（時間戳可以是系統給的，也可以是自定義的）。複雜事件則是由基本事件通過各種時間關係、邏輯關係所組成的事件。

WITHIN：是指輸出的複雜事件e的結束時間減去e的開始時間需<=視窗大小。只有滿足這個條件的複雜事件e才會被輸出。

FROM：為事件類指定資料流來源。例如： FROM T1:stockstream,T2:pos 其中T1是股價流stockstream的別名，T2是pos交易流的別名。而stockstream、pos流可以看做兩張無限長的關係表。表包含的欄位是在輸入介面卡定義的。從2.0.23(sp1)版本開始，別名和流名稱可以空格分隔，如FROM stockstream T1,streamname alias等價於FROM T1:stockstream,alias:streamname。

SELECT：被查詢欄位或聚合函式、自定義函式。如果設定為*,則獲取整個POJO明細。另外，輸出介面卡的isOutputAsSelection設定為false，也會讓SELECT不起作用，從而獲取整個POJO明細。這時，通常需自定義輸出介面卡來選取或計算資料。

簡單過濾（Filter）查詢PATTERN直接寫事件類名，不需要複雜事件操作符。如電壓監測中，選取資料作實時圖形顯示。

CREATE QUERY VD0001 
SELECT T1.lineid AS fusionwidgetsid,T1.lineid AS lineid,T1.voltagevalue AS voltagevalue 
FROM T1:voltagestream 
PATTERN T1
WHERE T1.lineid='110kvline' 
WITHIN 0

WITHIN 設定為0，是因為Filter通常不需要視窗，每條基本事件資料按條件過濾。這條語句連線到圖形顯示輸出介面卡就可以實時顯示線路'110kvline'的電壓圖形了。

FROM T1:voltagestream 的T1為事件類名，voltagestream是資料流名。類似關係資料庫，也可以這樣理解，voltagestream流是一張無限長的關係表，T1則是voltagestream的別名。下文中讀者會發現，同一個流有多個別名，多個別名在一個流上來構建複雜模式。

1. 複雜事件基本操作符

1.1與模式（都發生）

Conjunction(A&B):表示事件A、事件B都發生，但不規定A、B發生的順序。

例：IT系統運維監控中，“服務呼叫開始”觸發一個事件，在“呼叫結束”觸發一個事件，如果此呼叫的處理時間沒超時(1000ms)的話，就輸出服務處理時間

CREATE QUERY callnottimeout
SELECT T2._start_time_-T1._start_time_ AS responsetime, T1.callerid AS functionname ,'false' AS timeout
FROM T1:callstream,T2:callstream
PATTERN T1&T2
WHERE T1.callerid=T2.callerid  AND T1.eventtype ='start' AND T2.eventtype ='end'
WITHIN 1000

1.2順序發生

Sequence(A;B): 表示事件A在事件B後發生，即事件A的結束時間小於事件B的開始時間。

例：當RFID Reader 讀到標籤後，在1s內該標籤還在則認為是連續讀到該標籤，如果2s後該標籤還存在，則觸發進入通道事件

CREATE QUERY readenterchannel
SELECT 'enterchannel' AS type,T3.id AS T3_id,T3.num AS num
FROM T1:readerevent,T2:readerevent,T3:readerevent
PATTERN T1;T2;T3 
WHERE T1.id=T2.id AND T1.id=T3.id AND T2._end_time_-T1._end_time_<=1000 AND T3._end_time_-T2._end_time_<2000 AND T3._end_time_-T1._end_time_>1000  
WITHIN 3000

1.3 非模式

Negation（!A）:表示事件A不發生。非模式必須在兩個基本事件之間使用。如“A;!B;C”表示A、C順序發生，但其間不能有B發生。用在超時監測，事務審計等場景中比較多。

例：當RFID Reader 讀到標籤後，2秒內沒有再讀到標籤，則認為標籤離開了通道

CREATE QUERY readoutofchannel2
SELECT T1.id AS T1_id,T1.num AS T1_num,'outofchannel' AS type
FROM T1:readerevent,T2:readerevent,T3:delay2sectimer
PATTERN T1;!T2;T3 
WHERE T1.id=T3.id AND T2.id=T3.id AND T3._end_time_-T1._end_time_=2000 
WITHIN 2000

為了保證結果正確性，SODBASE CEP引擎規定條件中"非事件"T2不能與它之前的事件T1做關聯，而要與T3關聯，即不能有類似T1.id=T2.id的條件。

例：IT系統運維監控中，“服務呼叫開始”觸發一個事件，在“呼叫結束”觸發一個事件，如果超時(1000ms)的話，就輸出超時事件

CREATE QUERY calltimeoutnotification 
SELECT '-1' AS responsetime, T1.callerid AS functionname,'true' AS timeout 
FROM T1:callstream,T2:callstream,T3:calltimeoutevent 
PATTERN T1;!T2;T3  
WHERE T3._end_time_-T1._end_time_=1000 AND T2.callerid=T3.callerid AND T1.eventtype='start' 
WITHIN 1000

1.4或模式

Disjunction（A|B）:表示事件A發生或事件B發生或兩者都發生。

例：變電站監測中，查詢220KV I段PT電壓或其它段電壓>112的事件

CREATE QUERY VD0000_1 
SELECT * 
FROM T1:VD0000_1.模擬電壓,T2:VD0000_1.模擬電壓 
PATTERN T1|T2 
WHERE T1.lineid='220KV I段PT電壓' 
AND T2.voltagevalue>112 
WITHIN 500

1.5克林包（Kleen Closure）

Kleen Closure(A^+/A^num): A^+表示事件A發生1次或多次；A^num中num為數字，表示事件A發生num次。kleen closure必須在兩個基本事件之間使用。例如: A;B^+;C表示A發生後，B發生多次，然後C發生

還有一個A^*,表示A發生0次或多次。但是A發生0次時取A的欄位值是空值，使用時要注意，測試保證能取到結果。否則請使用(A;C)|(A;B^+;C)來代替A;B^*;C。

1.5.1 普通視窗查詢

例：金融風控中，查詢1小時內的交易大於10000且交易次數大於2筆的卡號

CREATE QUERY pos
SELECT T1.acctnum AS acctum,T1.value AS T1_value,tostring(T2.value) AS T2.value,T3.value AS T3_value
FROM T1:pos,T2:pos,T3:pos
PATTERN T1;T2^+;T3
WHERE T1. acctnum =T2. acctnum
AND T3. acctnum =T1. acctnum
AND T1.value+T3.value+sum(T2.value)>10000
WITHIN 3600000

1.5.2定時統計

例：查詢股價的10秒鐘K線資料

CREATE QUERY vwap1 
SELECT min(T2.price) AS LOW, max(T2.price) AS HIGH 
FROM T1:timer,T2:stock,T3:timer 
PATTERN T1;T2^+;T3  
WHERE T1._start_time_=T3._start_time_-10000  
WITHIN 10000

average是內建聚合函式，還有sum、tostring、max、min、first、count、countdistinct、tostringdistinct等。timer流是用定時觸發輸入介面卡生成的，週期為10000ms。相當於每個10秒視窗被T1和T3這兩個定時觸發時間點給括起來了。

例：整點開始統計12:00:00~12:00:10,12:00:10~12:00:20等每10秒的股票資料

很多情況下，我們還想將每個視窗的起始時間設為整點開始，按一定的滑動視窗統計資料。

一種方法是用帶起始時間的定時觸發器，設定定時器開始時間，如2014-01-01 12:00:00，週期(ms)為10000。在sodbase cep 2.0.20(sp1)之後版本中如果開始時間在當前時間之前，會按照週期調整到當前時間之後的觸發點。

另外，前例語句中T1和T3將T2括起來是開區間視窗，即T1._end_time_<T2._start_time<=T2._end_time<T3._start_time_。如果要精確的包含timer時間點上的資料，可以再用一個介面卡：延時介面卡。將timer1通過延時介面卡延時1ms輸出為timer2流。也就是timer1和timer2起始時間相差一個單位時間。最終的EPL語句為

CREATE QUERY stockquery 
SELECT min(T2.price) AS LOW, max(T2.price) AS HIGH 
FROM T1:timer1,T2:stock,T3:timer2 
PATTERN T1;T2^+;T3  
WHERE T1._start_time_=T3._start_time_-10001  
WITHIN 10001

1.5.3 視窗內資料聚類(GROUP BY)

針對kleen closure進行資料分組。GROUP BY需緊接著PATTERN寫。

例：統計每一隻股票的10s的統計值,2s輸出一次

timer為定時輸入介面卡，週期為2s

CREATE QUERY demo 
SELECT average(T2.price) AS price_avg, T2.name AS name  
FROM T1:timer,T2:stock,T3:timer 
PATTERN T1;T2^+;T3  
GROUP BY T2.name
WHERE T1._start_time_=T3._start_time_-10000  
WITHIN 10000

例：網路流量分析中，統計10s內從源地址到目的地址的package數量

SELECT T2.src AS src,.destination AS destination, count(T2.src) AS count 
FROM T1:timer1,T2:netstream,T3:timer2 
PATTERN T1;T2^+;T3  
GROUP BY T2.src,T2.destination  
WHERE T3._start_time_-T1._start_time_=10001  
WITHIN 10001

netstream是流量包資訊，包含源地址src、目的地址destination、通訊協議、包大小等資料。timer1、timer2是相差1ms的定時器，週期都是10s

1.5.4 事件觸發的視窗開啟

例：每當股價超過閾值(>50.0)後，開啟10秒視窗，統計10秒內股價超過閾值的事件,這10秒內的超過閾值事件，就不再開啟新的10秒視窗

CREATE QUERY delayoutput2 
SELECT * 
FROM T1:delayoutput.stock,T2:delayoutput.stock,T3:delayoutput2.intervaltimer 
PATTERN T1;T2^*;T3  
WHERE T1._start_time_=T3._start_time_-10000  AND T1.price>50 AND T2.price>50  
WITHIN 10000 
BATCHMODE

流delayoutput2.intervaltimer中是每個股價資料延時10s生成的事件，與delayoutput.stock中的事件一一對應形成10秒視窗，參見下文延時輸出介面卡

BATCHMODE關鍵詞表示檢測到結果後，滑動視窗的開始時間移到之前結果的結束時間後。即下一個輸出結果的T1._start_time_>=前一個結果的T3._end_time_。

類似的例子還有

例：電信裝置監測中發現故障後，開啟視窗記錄裝置詳細資訊。

例：車輛行駛記錄發現異常後，開啟視窗監測車輛資料做記錄。

1.5.5 查詢視窗內某欄位最大值對應的事件資訊

例：在IT裝置監測中，監測CPU使用率最大的機器的ID,滑動視窗為10秒

 CREATE QUERY maxCPUMachine 
 SELECT JAVASTATIC:f.Top:getByIndex(tostring(T2.machineId),JAVASTATIC:f.Top:indexOfMax(tostring(T2.cpu_rate))) AS max_machineId,
        max(T2.cpu_rate) AS max_cpu_rate 
   FROM T1:timer1,T2:cpustream,T3:timer1 
PATTERN T1;T2^+;T3  
 WITHIN 10000

注：如果是一張靜態的傳統關係型資料庫表，這類查詢通常需要先查出最大值，然後與原表進行Join得到CPU使用率最大的對應裝置。在這裡，我們直接使用自定義函式去取CPU使用率最大的對應裝置ID了。在流式資料中，做Topk查詢也可以採用類似思路，或者自定義輸出介面卡來處理複雜的聚合運算。

1.5.6 標記視窗內發生某事件

例：5秒一個視窗，視窗批次滑動，查視窗內某裝置的CPU使用率>0.8

   SELECT JAVASTATIC:f.Math:i_divide(T1._start_time_,'5000') AS timeinterval,T2.machine_id 
    FROM  T1:timer1,T2:nextstream,T3:timer2 
 PATTERN  T1;T2^+;T3  
GROUP BY  T2.machine_id
   WHERE  T3._start_time_-T1._start_time_=5001  AND T2.cpu_rate>0.8
  WITHIN  5001

注：i_divide為自定義函式做整數除法。

1.5.7 非嚴格順序的Kleen Closure視窗(EXTENDED關鍵字)

前文講到的A;B^*;C預設B要在A之後發生，C要在B之後發生。

使用EXTENDED關鍵字後，A._end_time_=B._start_time_,B._end_time_=C._start_time_複雜事件也會被輸出。使用此關鍵字，系統只保證相同時間戳的資料全部輸入後，才能夠監測到所有情況。

1.6 使用括號

可以使用括號構造更為複雜的模式，在實際應用中有時也會使用到。

例如：(T1;T2^2;T3)|(T4;T5)

當然，為了使模型拓撲更清晰和單元測試更方便，建議是用級聯方式（見級聯輸入輸出介面卡）構建CEP模型，而不是寫非常複雜的PATTERN表示式。上例中，可以用一個單元模型(T1;T2^2;T3)輸出級聯到流stream1,另一個單元模型T4;T5輸出級聯到stream2。再用一個單元模型T6|T7 FROM T6:stream1,T7:stream2得到最終結果。

2.自定義函式和內建函式

2.1 自定義函式

在EPL的SELECT和WHERE語句中都可以方便地呼叫使用者自己寫的Java函式

例：IT系統運維監控中，查詢服務呼叫和子服務呼叫之間的時間差

CREATE QUERY calltime
SELECT A.callerid,B.callerid,JAVA:com.example.CallAnalysis:minus(A.time,B.time) AS timecost 
FROM A:callstream,B:callstream 
PATTERN A;B  
WHERE JAVA:com.example.CallAnalysis:parentOf(A.callerid,B.callerid)   
WITHIN 6000

寫Java函式時，引數需要是String型別，返回需要是String,Double,Integer,Float。

在最新的SODBASE CEP引擎中以JAVA開頭呼叫物件方法，以JAVASTATIC開頭呼叫類靜態方法。

例：IT系統監控中，CPU利用率超過80%,報警一次。接下來的2min中，如果CPU利用率還超80%，也不報警。

CREATE QUERY reduceAlarmNum
SELECT 'CPU_HIGH' AS event.type, JAVA:package.class:setLastAlarmTime('CPU_HIGH',event._start_time_) AS setalarmstate 
FROM event:cpuusagestream
PATTERN event
WHERE event.cpu_usage>0.8 AND event._start_time_-JAVA:package.class:getLastAlarmTime('CPU_HIGH')>12000
WITHIN 0

這裡的JAVA:package.class:getLastAlarmTime('CPU_HIGH')和JAVA:package.class:setLastAlarmTime('CPU_HIGH',event._start_time_) 是自定義Java函式，作用是在全域性變數或快取、儲存系統中記錄上次同類報警的時間。

2.2 內建函式

內建函式是為了方便使用，引擎自帶一些內建函式，內建函式的字母均為小寫。如前文提到的克林包上的average、sum、tostring、max、min、first、count、countdistinct、tostringdistinct函式。新版本中還支援下面幾個常用函式。

字串連線函式：concat

布林運算函式：and、or、xor

為了避免函式巢狀的型別匹配不正確，建議儘量使用自定義函式。

3.常用輸入輸出介面卡

要在實際專案中用好EPL,知道常用的輸入輸出介面卡是必不可少的，因為現實模型往往不是一條EPL語句就可以建模的，而是多條EPL連起來，多種流、多種操作連起來的。

3.1宣告輸入（程式輸入或級聯輸入）

com.sodbase.inputadaptor.StubInputAdaptor

引數1：流名稱

本身不產生資料，宣告資料要從程式中輸入或從其它EPL的輸出級聯輸入。

3.2定時、延時事件輸出

3.2.1定時事件生成

com.sodbase.inputadaptor.timer.TimerInputAdaptor

引數1：流名稱

引數2：開始時間，需要在當前時間後格式"yyyy-MM-dd HH:mm:ss"

引數3：週期

還有一個沒有開始時間的定時事件

com.sodbase.inputadaptor.TimerInputAdaptor

引數1：流名稱

引數2：週期

3.2.2定時延時

com.sodbase.outputadaptor.timer.FixedDelayTimerOutputAdaptor

引數1：延時事件生成的流名稱

引數2：延時時間

例：為callstream中的eventtype ='start'的事件生成延時事件，形成新的流calltimeoutevent

CREATE QUERY calltimeout
SELECT 'timer' AS type,T1.time AS time,T1.callerid AS callerid
FROM T1:callstream
PATTERN T1
WHERE T1.eventtype ='start'
WITHIN 0

輸出介面卡配置

<outputAdaptors>
        <isOutputAsSelection>true</isOutputAsSelection>
        <outputAdaptorClassName>com.sodbase.outputadaptor.timer.FixedDelayTimerOutputAdaptor</outputAdaptorClassName>
        <adaptorParams>calltimeoutevent</adaptorParams><!--超時事件流名稱為calltimeoutevent-->
        <adaptorParams>1000</adaptorParams><!--超時閾值為1000ms，即超時事件在呼叫開始後1s時發生-->
        <isExternal>false</isExternal><!--預設false，嵌入式開發沒有多使用者時一般用不到-->
        <queryName>calltimeout</queryName><!--此輸出介面卡屬於名為calltimeout的EPL-->
</outputAdaptors><span style="font-size:18px;">
</span>

3.2.3變數延時

com.sodbase.outputadaptor.timer.DelayTimerOutputAdaptor

引數1：延時事件生成的流名稱

引數2：確定延時時間的欄位，即延時長短由事件中的欄位值決定

在資訊系統中做事件引擎可用於流程審批使用者自定義超時跳轉，在演算法交易中可用於定時交易或拆單策略

3.3 螢幕列印輸出

com.sodbase.outputadaptor.PrintEventOutputAdaptor

引數：無

用於將事件列印在螢幕上，一般在單元測試和除錯中使用

3.4 一個事件生成多個事件

com.sodbase.outputadaptor.eventsplit.EventSplitOutputAdaptor

引數1： inputStreamConnected，生成的流名稱
引數2： spawnnumber : 一個事件生成多少個事件
引數3： idfieldname : 生成的事件中哪個欄位使用者編號（0..n）
引數4： retainFields : true or false ，是否保留原事件的欄位

在演算法交易中可用於拆單策略，定時資料處理中，可用於資料拆分處理。

3.5 級聯輸出介面卡

即一個EPL的輸出作為另一個EPL的輸入

3.5.1 不帶watermark級聯輸出

com.sodbase.outputadaptor.connection.ConnectToSodInputOutputAdaptor

作用：所有事件按原樣輸出，不做watermark過濾

引數1：級聯到的流名稱

3.5.2 帶watermark級聯輸出

帶watermark的級聯輸出可以保證只執行一次，亂序容錯，watermark值以時間戳的單位時間來衡量，設定的越大，對於亂序和重複事件的容忍度越大，

com.sodbase.outputadaptor.ConnectToSodInputOutputAdaptor