本文根據Apache Flink 實戰&進階篇系列直播課程整理而成，由哈囉出行大資料實時平臺資深開發劉博分享。通過一些簡單的實際例子，從概念原理，到如何使用，再到功能的擴充套件，希望能夠給打算使用或者已經使用的同學一些幫助。

主要的內容分為如下三個部分：

1. Flink CEP概念以及使用場景。
2. 如何使用Flink CEP。
3. 如何擴充套件Flink CEP。

Flink CEP 概念以及使用場景

什麼是 CEP

CEP的意思是複雜事件處理，例如：起床-->洗漱-->吃飯-->上班等一系列串聯起來的事件流形成的模式稱為CEP。如果發現某一次起床後沒有刷牙洗臉亦或是吃飯就直接上班，就可以把這種非正常的事件流匹配出來進行分析，看看今天是不是起晚了。

下圖中列出了幾個例子：

• 第一個是異常行為檢測的例子：假設車輛維修的場景中，當一輛車出現故障時，這輛車會被送往維修點維修，然後被重新投放到市場執行。如果這輛車被投放到市場之後還未被使用就又被報障了，那麼就有可能之前的維修是無效的。
• 第二個是策略營銷的例子：假設打車的場景中，使用者在APP上規劃了一個行程訂單，如果這個行程在下單之後超過一定的時間還沒有被司機接單的話，那麼就需要將這個訂單輸出到下游做相關的策略調整。
• 第三個是運維監控的例子：通常運維會監控伺服器的CPU、網路IO等指標超過閾值時產生相應的告警。但是在實際使用中，後臺服務的重啟、網路抖動等情況都會造成瞬間的流量毛刺，對非關鍵鏈路可以忽略這些毛刺而只對頻繁發生的異常進行告警以減少誤報。

Flink CEP 應用場景

• 風險控制：對使用者異常行為模式進行實時檢測，當一個使用者發生了不該發生的行為，判定這個使用者是不是有違規操作的嫌疑。
• 策略營銷：用預先定義好的規則對使用者的行為軌跡進行實時跟蹤，對行為軌跡匹配預定義規則的使用者實時傳送相應策略的推廣。
• 運維監控：靈活配置多指標、多依賴來實現更復雜的監控模式。

Flink CEP原理

Flink CEP內部是用NFA（非確定有限自動機）來實現的，由點和邊組成的一個狀態圖，以一個初始狀態作為起點，經過一系列的中間狀態，達到終態。點分為起始狀態、中間狀態、最終狀態三種，邊分為take、ignore、proceed三種。

• take
  ：必須存在一個條件判斷，當到來的訊息滿足take邊條件判斷時，把這個訊息放入結果集，將狀態轉移到下一狀態。
• ignore：當訊息到來時，可以忽略這個訊息，將狀態自旋在當前不變，是一個自己到自己的狀態轉移。
• proceed：又叫做狀態的空轉移，當前狀態可以不依賴於訊息到來而直接轉移到下一狀態。舉個例子，當用戶購買商品時，如果購買前有一個諮詢客服的行為，需要把諮詢客服行為和購買行為兩個訊息一起放到結果集中向下遊輸出；如果購買前沒有諮詢客服的行為，只需把購買行為放到結果集中向下遊輸出就可以了。 也就是說，如果有諮詢客服的行為，就存在諮詢客服狀態的上的訊息儲存，如果沒有諮詢客服的行為，就不存在諮詢客服狀態的上的訊息儲存，諮詢客服狀態是由一條proceed邊和下游的購買狀態相連。

下面以一個打車的例子來展示狀態是如何流轉的，規則見下圖所示。

以乘客制定行程作為開始，匹配乘客的下單事件，如果這個訂單超時還沒有被司機接單的話，就把行程事件和下單事件作為結果集往下游輸出。

假如訊息到來順序為：行程-->其他-->下單-->其他。

狀態流轉如下：

1、開始時狀態處於行程狀態，即等待使用者制定行程。

2、當收到行程事件時，匹配行程狀態的條件，把行程事件放到結果集中，通過take邊將狀態往下轉移到下單狀態。

3、由於下單狀態上有一條ignore邊，所以可以忽略收到的其他事件，直到收到下單事件時將其匹配，放入結果集中，並且將當前狀態往下轉移到超時未接單狀態。這時候結果集當中有兩個事件：制定行程事件和下單事件。

4、超時未接單狀態時，如果來了一些其他事件，同樣可以被ignore邊忽略，直到超時事件的觸發，將狀態往下轉移到最終狀態，這時候整個模式匹配成功，最終將結果集中的制定行程事件和下單事件輸出到下游。

上面是一個匹配成功的例子，如果是不成功的例子會怎麼樣？

假如當狀態處於超時未接單狀態時，收到了一個接單事件，那麼就不符合超時未被接單的觸發條件，此時整個模式匹配失敗，之前放入結果集中的行程事件和下單事件會被清理。

Flink CEP程式開發

本節將詳細介紹Flink CEP的程式結構以及API。

Flink CEP 程式結構

主要分為兩部分：定義事件模式和匹配結果處理。

官方示例如下：

DataStream<Event> input = ...
Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(
        new SimpleCondition<Event>() {
            @Override
            public boolean filter(Event event) {
                return event.getId() == 42;
            }
        }
    ).next("middle").subtype(SubEvent.class).where(
        new SimpleCondition<SubEvent>() {
            @Override
            public boolean filter(SubEvent subEvent) {
                return subEvent.getVolume() >= 10.0;
            }
        }
    ).followedBy("end").where(
         new SimpleCondition<Event>() {
            @Override
            public boolean filter(Event event) {
                return event.getName().equals("end");
            }
         }
    );

PatternStream<Event> patternStream = CEP.pattern(input, pattern);

DataStream<Alert> result = patternStream.select(
    new PatternProcessFunction<Event, Alert>() {
        @Override
        public void select(
                Map<String, List<Event>> pattern,
                Context ctx,
                Collector<Alert> out) throws Exception {
            out.collect(createAlertFrom(pattern));
        }
    });

程式結構分為三部分：首先需要定義一個模式(Pattern)，即第2行程式碼所示，接著把定義好的模式繫結在DataStream上（第25行），最後就可以在具有CEP功能的DataStream上將匹配的結果進行處理（第27行）。

下面對關鍵部分做詳細講解：

定義模式：上面示例中，分為了三步，首先匹配一個ID為42的事件，接著匹配一個體積大於等於10的事件，最後等待收到一個name等於end的事件。
匹配結果輸出：此部分，需要重點注意select函式（第30行，注：本文基於Flink 1.7版本）裡邊的Map型別的pattern引數，Key是一個pattern的name，它的取值是模式定義中的Begin節點start，或者是接下來next裡面的middle，或者是第三個步驟的end。後面的map中的value是每一步發生的匹配事件。因在每一步中是可以使用迴圈屬性的，可以匹配發生多次，所以map中的value是匹配發生多次的所有事件的一個集合。

Flink CEP構成

上圖中，藍色方框代表的是一個個單獨的模式；淺黃色的橢圓代表的是這個模式上可以新增的屬性，包括模式可以發生的迴圈次數，或者這個模式是貪婪的還是可選的；橘色的橢圓代表的是模式間的關係，定義了多個模式之間是怎麼樣串聯起來的。通過定義模式，新增相應的屬性，將多個模式串聯起來三步，就可以構成了一個完整的Flink CEP程式。

定義模式

下面是示例程式碼：

pattern.next("start").where(
        new SimpleCondition<Event>() {
            @Override
            public boolean filter(Event event) {
                return event.getId() == 42;
            }
        }
)

定義模式主要有如下5個部分組成：

pattern：前一個模式
next/followedBy/...：開始一個新的模式
start：模式名稱
where：模式的內容
filter：核心處理邏輯

模式的屬性

接下來介紹一下怎樣設定模式的屬性。模式的屬性主要分為迴圈屬性和可選屬性。

迴圈屬性可以定義模式匹配發生固定次數（times），匹配發生一次以上（oneOrMore），匹配發生多次以上。(timesOrMore)。

可選屬性可以設定模式是貪婪的（greedy），即匹配最長的串，或設定為可選的（optional），有則匹配，無則忽略。

模式的有效期

由於模式的匹配事件存放在狀態中進行管理，所以需要設定一個全域性的有效期（within）。 若不指定有效期，匹配事件會一直儲存在狀態中不會被清除。至於有效期能開多大，要依據具體使用場景和資料量來衡量，關鍵要看匹配的事件有多少，隨著匹配的事件增多，新到達的訊息遍歷之前的匹配事件會增加CPU、記憶體的消耗，並且隨著狀態變大，資料傾斜也會越來越嚴重。

模式間的聯絡

主要分為三種：嚴格連續性（next/notNext），寬鬆連續性（followedBy/notFollowedBy），和非確定寬鬆連續性（followedByAny）。

三種模式匹配的差別見下表所示：

總結如下：

• 嚴格連續性：需要訊息的順序到達與模式完全一致。
• 寬鬆連續性：允許忽略不匹配的事件。
• 非確定寬鬆連性：不僅可以忽略不匹配的事件，也可以忽略已經匹配的事件。

多模式組合

除了前面提到的模式定義和模式間的聯絡，還可以把相連的多個模式組合在一起看成一個模式組，類似於檢視，可以在這個模式檢視上進行相關操作。

上圖這個例子裡面，首先匹配了一個登入事件，然後接下來匹配瀏覽，下單，購買二手手機號拍賣平臺地圖這三個事件反覆發生三次的使用者。

如果沒有模式組的話，程式碼裡面瀏覽，下單，購買要寫三次。有了模式組，只需把瀏覽，下單，購買這三個事件當做一個模式組，把相應的屬性加上times(3)就可以了。

處理結果

處理匹配的結果主要有四個介面： PatternFlatSelectFunction，PatternSelectFunction，PatternFlatTimeoutFunction和PatternTimeoutFunction。

從名字上可以看出，輸出可以分為兩類：select和flatSelect指定輸出一條還是多條，timeoutFunction和不帶timeout的Function指定可不可以對超時事件進行旁路輸出。

下圖是輸出的綜合示例程式碼：

狀態儲存優化

當一個事件到來時，如果這個事件同時符合多個輸出的結果集，那麼這個事件是如何儲存的？

Flink CEP通過Dewey計數法在多個結果集中共享同一個事件副本，以實現對事件副本進行資源共享。

Flink CEP的擴充套件

本章主要介紹一些Flink CEP的擴充套件，講述如何做到超時機制的精確管理，以及規則的動態載入與更新。

超時觸發機制擴充套件

原生Flink CEP中超時觸發的功能可以通過within+outputtag結合來實現，但是在複雜的場景下處理存在問題，如下圖所示，在下單事件後還有一個預付款事件，想要得到下單並且預付款後超時未被接單的訂單，該如何表示呢？

參照下單後超時未被接單的做法，把下單並且預付款後超時未被接單規則表示為下單.followedBy(預付款).followedBy(接單).within(time)，那麼這樣實現會存在問題嗎？

這種做法的計算結果是會存在髒資料的，因為這個規則不僅匹配到了下單並且預付款後超時未被接單的訂單（想要的結果），同樣還匹配到了只有下單行為後超時未被接單的訂單（髒資料，沒有預付款）。原因是因為超時within是控制在整個規則上，而不是某一個狀態節點上，所以不論當前的狀態是處在哪個狀態節點，超時後都會被旁路輸出。

那麼就需要考慮能否通過時間來直接對狀態轉移做到精確的控制，而不是通過規則超時這種曲線救國的方式。 於是乎，在通過訊息觸發狀態的轉移之外，需要增加通過時間觸發狀態的轉移的支援。要實現此功能，需要在原來的狀態以及狀態轉移中，增加時間屬性的概念。如下圖所示，通過wait運算元來得到waiting狀態，然後在waiting狀態上設定一個十秒的時間屬性以定義一個十秒的時間視窗。

wait運算元對應NFA中的ignore狀態，將在沒有到達時間視窗結束時間時自旋，在ComputationState中記錄wait的開始時間，在NFA的doProcess中，將到來的資料與waiting狀態處理，如果到了waiting的結束時間，則進行狀態轉移。

上圖中紅色方框中為waiting狀態設定了兩條ignore邊：

1.waitingStatus.addIgnore(lastSink,waitingCondition)，waitingCondition中的邏輯是獲取當前的時間（支援事件時間），判斷有沒有超過設定的waiting閾值，如果超過就把狀態向後轉移。
2.waitingStatus.addIgnore(waitingCondition)，waitingCondition中如果未達到設定的waiting閾值，就會自旋在當前的waiting狀態不變。

規則動態注入

線上執行的CEP中肯定經常遇到規則變更的情況，如果每次變更時都將任務重啟、重新發布是非常不優雅的。尤其在營銷或者風控這種對實時性要求比較高的場景，如果規則視窗過長（一兩個星期），狀態過大，就會導致重啟時間延長，期間就會造成一些想要處理的異常行為不能及時發現。

那麼要怎麼樣做到規則的動態更新和載入呢？

梳理一下整體架構，Flink CEP是執行在Flink Job裡的，而規則庫是放在外部儲存中的。首先，需要在執行的Job中能及時發現外部儲存中規則的變化，即需要在Job中提供訪問外部庫的能力。 其次，需要將規則庫中變更的規則動態載入到CEP中，即把外部規則的描述解析成Flink CEP所能識別的pattern結構體。最後，把生成的pattern轉化成NFA，替換歷史NFA，這樣對新到來的訊息，就會使用新的規則進行匹配。

下圖就是一個支援將外部規則動態注入、更新的介面。

這個接口裡面主要實現了四個方法：

• initialize：初始化方法，進行外部庫連線的初始化。
• inject：和外部資料庫互動的主要方法，監聽外部庫變化，獲取最新的規則並通過Groovy動態載入，返回pattern。
• getPeriod：設定輪巡週期，在一些比較簡單的實時性要求不高的場景，可以採用輪巡的方式，定期對外部資料庫進行檢測。
• getNfaKeySelector：和動態更新無關，用來支援一個流對應多個規則組。

歷史匹配結果清理

新規則動態載入到Flink CEP的Job中，替換掉原來的NFA之後，還需要對歷史匹配的結果集進行清理。在AbstractKeyedCEPPatternOperator中實現重新整理NFA，注意，歷史狀態是否需要清理和業務相關：

1. 修改的邏輯對規則中事件的匹配沒有影響，保留歷史結果集中的狀態。
2. 修改的邏輯影響到了之前匹配的部分，需要將之前匹配的結果集中的狀態資料清除，防止錯誤的輸出。

總結

使用Flink CEP，熟知其原理是很重要的，特別是NFA的狀態轉移流程，然後再去看原始碼中的狀態圖的構建就會很清晰了。