Flink CEP的論文與設計

Flink的CEP設計與實現重度參考了論文《Efficient Pattern Matching over Event Streams》。下面我們就來結合論文談談Flink CEP的設計。

這篇論文探討的話題是如何在事件流上進行高效地模式匹配。談及模式匹配，為大眾所知的可能是正則表示式匹配，而在流上運用正則表示式進行模式匹配有兩個挑戰：

• 要求豐富的語言特性：在事件流上進行模式匹配的語言明顯要比用正則表示式進行模式匹配的語言所需要的能力豐富得多。這些事件模式語言需要包含對錶達序列、Kleene閉包、否定以及複雜斷言的構建，同時還包含從混雜著相關、不相關事件的輸入流中提取相關事件的策略；
• 流上處理的效率：在事件流上進行的模式查詢如何被高效地計算，需要新的演算法和優化工作；

而這篇論文提出解決方案是：設計並實現了一個正式的計算模型：

NFAb，它包含一個非確定性的有限自動機（NFA）和一個匹配緩衝區（buffer）。該模型為完整的事件模式查詢集合提供了清晰的語義，允許進行優化並且可產生能在事件流上執行的查詢計算計劃，同時設計了一個共享的基於版本的緩衝區來優化針對每次執行構建獨立匹配緩衝區所帶來的資源開銷。

除此之外，論文還分析了執行時複雜度、展示了執行時的演算法實現與優化

NFA-b模型

NFAb這一計算模型是由Flink CEP所參考的《Efficient Pattern Matching over Event Streams》論文提出的。NFA[^1]（nondeterministic finite automaton，全稱：非確定有限自動機）是對每個狀態和輸入符號對可以有多個可能的下一個狀態的有限狀態自動機。NFAb相較於NFA的改進是它配備了一個匹配緩衝區（buffer），用來作為模式的查詢、計算模型。

考慮下面這個來自論文中的股票交易業務中的模式：

PATTERN SEQ(Stock+ a[ ], Stock b)
WHERE
 skip_till_next_match(a[ ], b) {
        [symbol]
  and a[1].volume > 1000
  and a[i].price > avg(a[..i-1].price)
  and b.volume < 80% * a[a.LEN].volume }
WITHIN
 1
 
 hour

模式中的”[symbol]”表示分割槽處理。

上面的這個模式，描述了一個複雜的股票交易趨勢：在過去的一段時間內，股票交易量開始升高，但在一個週期之後，當價格增長或者保持相對穩定後，交易量將會暴跌。這個模式有兩個輸入：在股票事件上的一個“正閉包”，結果儲存於a[]中；一個分離的單一的股票事件，儲存在b中。作用在a[1]上的斷言指定了初始交易量，而作用在a[i]（i > 1）上的斷言要求其當前事件的價格超過之前被選擇事件的平均值，這樣的斷言會捕獲交易的價格增長趨勢。最後一個斷言將b跟a[a.LEN]進行比較，這裡a.LEN關聯著a[]中最後一個被選擇的事件，它會捕獲最終交易量的落差。

狀態和狀態轉換

狀態和轉移函式（可類比成是銜接狀態轉換的邊）是

NFAb的兩種基本組成要素，用於示例的股票模式的NFAb結構如下圖所示：

起始狀態a[1]，表示匹配過程的開始，它等待“正閉包”的事件輸入並選擇一個事件到匹配緩衝區中的a[1]單元。在下一個狀態a[i]，它會嘗試選擇另一個事件並放入緩衝區中的a[i]（i > 1）單元。接下來的狀態b表示匹配過程對於a[]已經完成了一個特定的匹配且已經準備好處理下一個模式輸入。而最終狀態F，則表示處理完成，它將建立一個模式匹配。

CEP程式碼中以State類表示狀態，其完整類圖如下：

從類圖可見，它主要封裝了狀態的名稱、型別以及跟其有關的狀態集合。StateType是列舉型別，列舉值如下：

public enum StateType {
    Start,         // NFA的起始狀態
    Final,         // NFA的終止狀態
    Normal         // 非起始非終止狀態的其他正常狀態
}

從模式的狀態圖中可看到每個狀態都關聯著一組邊，表示在狀態上可以發生的轉換動作。正如上圖所展示的那樣，首狀態有一個“begin”邊，每個a[i]狀態有一個“proceed”邊以及一個迴圈的“take”邊。每個狀態（除了起始狀態和終止狀態）都有一個迴圈的“ignore”邊。

NFAb 會將模式中的“WHERE”以及“WITHIN”查詢子句翻譯成相關的語法附加到對應的邊上，股票模式的各個邊的條件語法如下圖所示：

上面的這些轉換動作，在程式碼中通過一個名為StateTransitionAction的列舉類來表示：

public enum StateTransitionAction {
    TAKE,         //獲得當前事件並將它分配給新狀態
    IGNORE,       //忽略當前事件並做狀態轉換
    PROCEED       //做狀態轉換並保留當前狀態以為後續處理
}

總結而言，在CEP中以State表示上圖中的節點，以StateTransition表示上圖中的邊，也即狀態之間的轉化。所以這兩個物件之間是互相關聯的關係：

狀態將用於NFA中。

微信掃碼關注公眾號：Apache_Flink

QQ掃碼關注QQ群：Apache Flink學習交流群（123414680）