Realtime Compute for Apache Flink：複雜事件處理（CEP）語句

背景資訊

相較於Apache Flink的CEP SQL，Realtime ComputeFlink版在其基礎上進行了增強。例如，支援輸出逾時匹配、支援鬆散串連（followedBy）、支援指定事件之間的連續性等。關於Apache Flink CEP SQL的基本能力，詳情請參見Pattern Recognition。

使用限制

• 僅Realtime Compute引擎vvr-6.0.2-flink-1.15及以上版本支援CEP SQL擴充文法。

• 僅Realtime Compute引擎vvr-6.0.5-flink-1.15及以上版本支援組合模式和NO SKIP文法。

輸出逾時匹配

+----+------+------------------+
| id | type |          rowtime |
+----+------+------------------+
|  1 |    A | 2022-09-19 12:00 |
|  2 |    B | 2022-09-19 12:01 |
|  3 |    A | 2022-09-19 12:02 |
|  4 |    B | 2022-09-19 12:05 |
+----+------+------------------+

SELECT *
FROM MyTable MATCH_RECOGNIZE (
  ORDER BY rowtime
  MEASURES
    A.id AS aid,
    B.id AS bid,
    A.rowtime AS atime,
    B.rowtime AS btime
  PATTERN (A B) WITHIN INTERVAL '2' MINUTES
  DEFINE
    A AS type = 'A',
    B AS type = 'B'
) AS T

+-----+-----+------------------+------------------+
| aid | bid |            atime |            btime |
+-----+-----+------------------+------------------+
|   1 |   2 | 2022-09-19 12:00 | 2022-09-19 12:01 |
+-----+-----+------------------+------------------+

SELECT *
FROM MyTable MATCH_RECOGNIZE (
  ORDER BY rowtime
  MEASURES
    A.id AS aid,
    B.id AS bid,
    A.rowtime AS atime,
    B.rowtime AS btime
  ONE ROW PER MATCH SHOW TIMEOUT MATCHES
  PATTERN (A B) WITHIN INTERVAL '2' MINUTES
  DEFINE
    A AS type = 'A',
    B AS type = 'B'
) AS T

+-----+--------+------------------+------------------+
| aid |    bid |            atime |            btime |
+-----+--------+------------------+------------------+
|   1 |      2 | 2022-09-19 12:00 | 2022-09-19 12:01 |
|   3 | <NULL> | 2022-09-19 12:00 |           <NULL> |
+-----+--------+------------------+------------------+

事件之間的連續性

開源Flink CEP Java API中支援通過嚴格連續（next()）、鬆散連續（followedBy()）、非確定性鬆散連續（followedByAny()）、嚴格非連續（notNext()）、鬆散非連續（notFollowedBy()）等方式定義事件之間的連續性策略。

在Flink CEP SQL中預設使用的嚴格連續策略，即滿足模式比對的相鄰事件之間需要嚴格連續，連續的輸入事件之間不能存在非匹配的事件。例如上述例子中的模式 (A B) 指定A和B事件之間需要緊密相連。阿里雲Realtime ComputeFlink版對此進行了擴充，支援與Java API完全對等的表達能力。

{a1 b1}
{a3 b2}

{a1 b1}
{a2 b2}
{a3 b2}

{a1 b1}
{a2 b2}
{a3 b2}

{a2}

迴圈模式中的連續性以及貪婪匹配

開源Flink CEP Java API支援定義迴圈模式比對的連續性和貪婪策略，Apache Flink CEP SQL中預設使用嚴格連續且貪婪策略。例如A+匹配的多個A事件之間不允許有其他事件存在，並且匹配儘可能多的A事件。通過在迴圈量詞（例如*、+、{3, }）後增加?符號來指定連續性和貪婪策略。

{a2 a3 c1}
{a3 c1}

{a2 a3}
{a3 c1}

{a1 a2 a3 c1}
{a2 a3 c1}
{a3 c1}

{a1 a2 a3}
{a2 a3}
{a3 c1}

迴圈模式指定停止條件（Until）

開源Flink CEP Java API支援使用函數until(condition)來為迴圈模式指定其停止條件，其效果是在匹配迴圈模式的過程中，若當前事件符合until條件，則立刻終止當前迴圈模式的匹配，並從當前事件開始匹配後續模式。在阿里雲Realtime ComputeFlink版的SQL作業中可以在迴圈量詞如+、*、{3, }後使用{ CONDITION }文法來表達until語義。

例如針對事件序列a1, d1, a2, b1, a3, c1，有匹配條件DEFINE A AS A.type = 'a' OR A.type = 'b', B AS B.type = 'b', C AS C.type = 'c'，不同的模式對應的匹配序列如下表所示：

a2 b1 a3 c1
b1 a3 c1
a3 c1

a3 c1

a2 c1
a3 c1

a1 a2 c1
a2 c1
a3 c1

組合模式（Group Pattern）

開源 Flink CEP Java API支援組合模式（group pattern），將多個模式組合為一個整體用在next()、followedBy()和followedByAny()函數中，並支援整體的迴圈。在阿里雲Realtime ComputeFlink版的SQL作業中使用SQL標準中的文法(...)來定義組合模式，支援使用迴圈量詞如+、*、{3, }等。

例如對於模式PATTERN (A (B C*)+? D)，其中(B C*)為一個組合模式，並指定該組合迴圈出現一次以上，?表明為非貪婪匹配，其對應Java代碼如下：

Pattern.<String>begin("A").where(...)
  .next(
  	Pattern.<String>begin("B").where(...)
  		.next("C").where(...).oneOrMore().optional().greedy().consecutive())
  .oneOrMore().consecutive()
  .next("D").where(...)

匹配成功時MEASURES可以提取匹配結果中的某些部分資訊來作為結果的一部分。例如某組合模式比對到的結果為b1，b2 c1，b3 c2 c3，則MEASURES可以只擷取其中的一部分資訊進行輸出（下文中的SQL所示），如果只關注事件b，則通過FIRST(B.id)可以拿到第一組匹配結果的b，FIRST(B.id，1)可以拿到第二組結果的b，以此類推，輸出的結果為b1，b2，b3。

SELECT *
FROM MyTable MATCH_RECOGNIZE (
  ORDER BY rowtime
  MEASURES
    FIRST(B.id) AS b1_id,
    FIRST(B.id,1) AS b2_id,
    FIRST(B.id,2) AS b3_id
  PATTERN (A (B C*)+? D)
  DEFINE
    A AS type = 'A',
    B AS type = 'B',
    C AS type = 'C',
    D AS type = 'D'
) AS T

需要注意的是，組合模式與其前一個模式之間的連續性聲明作用於組合模式中的第一個模式而非整個組合模式。例如在PATTERN (A {- X*? -} (B C))中模式A和組合模式(B C)之間使用了followedBy關係，則實際為聲明了A和B之間的followedBy關係，其執行效果為A和(B C)之間可以存在若干不匹配B的事件，而非若干不匹配(B C)的事件。如序列a1 b1 d1 b2 c1對於該模式無輸出，因為b1出現後匹配過程立刻進入組合模式(B C)，而d1無法匹配C模式導致序列匹配失敗。

AFTER MATCH NO SKIP策略

Flink CEP Java API中After Match策略預設為NO_SKIP，CEP SQL中的預設策略則為SKIP_TO_NEXT_ROW。阿里雲Realtime ComputeFlink版擴充了SQL標準中的AFTER MATCH語句，可通過AFTER MATCH NO SKIP語句來聲明NO_SKIP策略，NO_SKIP策略在完成一條序列的匹配時，不會終止或丟棄其他已經開始的匹配過程。

NO_SKIP策略的一種常用情境為結合followedByAny跳過一些匹配事件進行更鬆散的匹配。例如對於序列a1 b1 b2 b3 c1，PATTERN (A {- X* -} B {- Y*? -} C)（等價於Pattern.begin("A").followedByAny("B").followedBy("C")）在使用預設的AFTER MATCH SKIP TO NEXT ROW時得到結果a1 b1 c1，因為當a1 b1 c1完成匹配時所有以a1開頭的序列都會被丟棄，而使用AFTER MATCH NO SKIP則能得到所有匹配序列a1 b1 c1、a1 b2 c1和a1 b3 c1。