如何使用Apache Flink CEP SQL从已经匹配的模式中获取事件？

问题描述

我的要求是基于2个事件（EVT_A和EVT_B与顺序无关）生成触发器。这是期望

1. EVT_A arrived. --> No action
2. EVT_B arrived  --> Should Trigger
3. EVT_B arrived  --> should Trigger since A was received prevIoUsly (o/p should include A and current B)
4. EVT_A arrived  --> should Trigger since B was received prevIoUsly (o/p should include current A and last B)
5. EVT_A arrived  --> should Trigger since B was received prevIoUsly (o/p should include current A and last B)

我尝试关注但没有成功。

SELECT E.*
From MyEvents
MATCH_RECOGNIZE (
    ORDER BY procTime
    MEASURES ARRAY[
        Event(A.id,A.name,A.date),Event(B.id,B.name,B.date)
    ] AS Events
    AFTER MATCH SKIP TO NEXT ROW
    PATTERN (A C* B)
    DEFINE
        A AS name in ('EVT_A','EVT_B'),B AS name in ('EVT_A','EVT_B') AND B.name <> A.name,C AS name not in ('EVT_A','EVT_B')
) AS E;

我还尝试了“ 第一次比赛后跳过”。但这也有例外。有关如何使用Flink sql CEP或Flink中其他任何方式实现此目标的任何建议。

解决方法

这似乎是RichFlatMap或ProcessFunction是最简单的方法。只是需要一点状态：

ValueState<Event> lastA;
ValueState<Event> lastB;

然后处理每个传入事件的逻辑是这样的：

if EVT_A
  store event as lastA
  emit lastB with this event unless lastB is null
if EVT_B
  store event as lastB
  emit lastA with this event unless lastA is null

如果您正在寻找有关使用Flink的托管状态的介绍，则有tutorial in the docs。