Realtime Compute for Apache Flink:Instructions CEP

Contexte

Par rapport au langage SQL CEP d'Apache Flink, les instructions CEP de Realtime Compute for Apache Flink offrent des fonctionnalités améliorées, telles que la sortie des événements correspondants qui n'arrivent pas dans un intervalle de temps spécifique, la contiguïté relaxée via l'utilisation de followedBy(), et la configuration du modèle de contiguïté entre les événements. Pour plus d'informations sur les fonctionnalités de base du langage SQL CEP d'Apache Flink, consultez Reconnaissance de modèles.

Limites

• Seule la version vvr-6.0.2-flink-1.15 ou ultérieure du moteur Realtime Compute for Apache Flink prend en charge la syntaxe SQL CEP étendue.

• Seule la version vvr-6.0.5-flink-1.15 ou ultérieure du moteur Realtime Compute for Apache Flink prend en charge les modèles de groupe et la syntaxe AFTER MATCH NO SKIP.

Sortie des événements correspondants qui n'arrivent pas dans un intervalle de temps spécifique

+----+------+------------------+
| 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> |
+-----+--------+------------------+------------------+

Modèles de contiguïté entre les événements

L'API Java CEP d'Apache Flink prend en charge les modèles de contiguïté suivants entre les événements : contiguïté stricte via next(), contiguïté relaxée via followedBy(), contiguïté relaxée non déterministe via followedByAny(), non-contiguïté stricte via notNext() et non-contiguïté relaxée via notFollowedBy().

Par défaut, le langage SQL CEP d'Apache Flink utilise la contiguïté stricte. Dans ce modèle, tous les événements correspondants doivent apparaître strictement les uns après les autres, sans événement non correspondant entre eux. Dans l'exemple précédent, le modèle (A B) impose que l'événement A et l'événement B se succèdent strictement. Realtime Compute for Apache Flink étend cette capacité pour prendre en charge une expression équivalente à celle de l'API Java.

{a1 b1}
{a3 b2}

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

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

{a2}

Contiguïté et mise en correspondance gloutonne dans les modèles en boucle

L'API Java CEP d'Apache Flink vous permet de spécifier les stratégies de contiguïté et de mise en correspondance gloutonne au sein des modèles en boucle. Par défaut, le langage SQL CEP d'Apache Flink utilise la contiguïté stricte et la mise en correspondance gloutonne. Par exemple, dans le modèle A+, aucun autre événement n'est autorisé entre plusieurs événements A, et le nombre d'événements A mis en correspondance est maximisé. Vous pouvez ajouter un ou plusieurs points d'interrogation (?) après le quantificateur de boucle, tel que *, + ou {3, }, afin de spécifier les stratégies de contiguïté et de mise en correspondance gloutonne.

{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(condition) dans les modèles en boucle

L'API Java CEP d'Apache Flink vous permet d'utiliser la fonction until(condition) pour spécifier une condition d'arrêt pour un modèle en boucle. Si l'événement actuel d'un modèle en boucle satisfait la condition spécifiée par la fonction until(condition), la mise en correspondance du modèle en boucle actuel est immédiatement interrompue et la mise en correspondance du modèle suivant commence à partir de l'événement actuel. Dans les déploiements SQL de Realtime Compute for Apache Flink, vous pouvez ajouter la syntaxe { CONDITION } à un quantificateur de boucle, tel que +, * et {3, }, pour exprimer la sémantique until.

Le tableau ci-dessous décrit les séquences correspondantes pour différents modèles, lorsque la séquence d'événements d'entrée est a1, d1, a2, b1, a3, c1 et que la condition est 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

Modèle de groupe

L'API Java CEP d'Apache Flink prend en charge les modèles de groupe. Dans un modèle de groupe, plusieurs modèles sont combinés et utilisés dans la fonction next(), followedBy() ou followedByAny(). Un modèle de groupe peut être répété dans son ensemble. Dans les déploiements SQL de Realtime Compute for Apache Flink, vous pouvez utiliser la syntaxe (...) de la norme SQL pour définir un modèle de groupe. Des quantificateurs de boucle, tels que +, * et {3, }, peuvent être utilisés.

Par exemple, dans le modèle PATTERN (A (B C*)+? D) , (B C*) est un modèle de groupe déclaré pour apparaître plusieurs fois. Le point d'interrogation (?) indique l'utilisation de la stratégie de mise en correspondance non gloutonne. Exemple de code Java :

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

La clause MEASURES définit le contenu inclus dans la sortie d'un modèle de groupe correspondant. Par exemple, si les séquences correspondantes sont b1, b2 c1 et b3 c2 c3 à chaque fois que le modèle de groupe spécifié est mis en correspondance, vous pouvez utiliser la clause MEASURES pour afficher uniquement une partie des résultats de mise en correspondance. Si vous souhaitez que seul l'événement b soit affiché en sortie, utilisez FIRST(B.id) pour obtenir l'événement b de la première séquence correspondante et FIRST(B.id,1) pour obtenir l'événement b de la deuxième séquence correspondante. Appliquez la même méthode pour obtenir l'événement b de la troisième séquence correspondante. Ainsi, la sortie du modèle de groupe sera b1, b2, b3. Exemple de code :

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

Notez que la continuité déclarée entre un modèle de groupe et son modèle précédent s'applique au premier modèle du groupe plutôt qu'à l'ensemble du modèle de groupe. Par exemple, dans le modèle PATTERN (A {- X*? -} (B C)) , followedBy est utilisé entre le modèle A et le modèle de groupe (B C). Cela déclare une continuité followedBy entre le modèle A et le modèle B. Dans ce cas, plusieurs événements ne correspondant pas au modèle B peuvent exister entre le modèle A et le modèle de groupe (B C), mais aucun événement ne correspondant pas au modèle de groupe (B C) ne peut exister. Si aucune sortie n'est générée pour le modèle PATTERN (A {- X*? -} (B C)) basé sur la séquence d'événements d'entrée a1 b1 d1 b2 c1, c'est parce que le processus de mise en correspondance entre immédiatement dans le modèle de groupe (B C) après l'apparition de b1, et d1 ne correspond pas au modèle C. Par conséquent, la mise en correspondance de la séquence échoue.

Stratégie AFTER MATCH NO SKIP

Dans l'API Java CEP d'Apache Flink, la stratégie AFTER MATCH par défaut est NO_SKIP. Dans le langage SQL CEP d'Apache Flink, la stratégie AFTER MATCH par défaut est SKIP_TO_NEXT_ROW. Realtime Compute for Apache Flink étend la clause AFTER MATCH de la norme SQL. Vous pouvez utiliser la clause AFTER MATCH NO SKIP pour déclarer la stratégie NO_SKIP. Lorsque la stratégie NO_SKIP est utilisée, les processus de mise en correspondance existants ne sont ni interrompus ni ignorés lorsque la mise en correspondance d'une séquence est terminée.

La stratégie NO_SKIP est généralement utilisée conjointement avec followedByAny pour ignorer certains événements correspondants afin d'assurer une continuité relaxée. Par exemple, si la séquence d'événements d'entrée est a1 b1 b2 b3 c1, la sortie pour le modèle PATTERN (A {- X* -} B {- Y*? -} C) est a1 b1 c1 lorsque la stratégie par défaut AFTER MATCH SKIP TO NEXT ROW est utilisée. Le modèle PATTERN (A {- X -} B {- Y? -} C) équivaut à Pattern.begin("A").followedByAny("B").followedBy("C"). En effet, toutes les séquences commençant par a1 sont ignorées lorsque la mise en correspondance pour a1 b1 c1 est terminée. Toutefois, si vous utilisez AFTER MATCH NO SKIP, toutes les séquences correspondantes peuvent être obtenues. Dans ce cas, a1 b1 c1, a1 b2 c1 et a1 b3 c1 sont renvoyés.