Flink CEP sql限制输出

Question

我有一个用例，其中我有两个输入主题的卡夫卡。主题模式：eventName, ingestion_time(will be used as watermark), orderType, orderCountry

第一个主题的数据：{"eventName": "orderCreated", "userId":123, "ingestionTime": "1665042169543", "orderType":"ecommerce","orderCountry": "UK"}

第二个主题的数据：{"eventName": "orderSucess", "userId":123, "ingestionTime": "1665042189543", "orderType":"ecommerce","orderCountry": "USA"}

我希望获得orderType，orderCountry的所有用户I，其中用户执行第一个事件，而不是在5分钟的窗口中获得第二个事件，orderType和orderCountry的每个用户最多有2个事件(即最多10分钟)。

我结合了两个主题数据，并在其之上创建了一个视图，并试图使用flink cep sql来获取输出，但不知何故无法弄清楚。

SELECT *
FROM union_event_table
    MATCH_RECOGNIZE(
        PARTITION BY orderType,orderCountry
        ORDER BY ingestion_time
        MEASURES
            A.userId as userId
            A.orderType as orderType
            A.orderCountry AS orderCountry
        ONE ROW PER MATCH
        PATTERN (A not followed B) WITHIN INTERVAL '5' MINUTES
        DEFINE
            A As A.eventName = 'orderCreated'
            B AS B.eventName = 'orderSucess'
    )

第一件事无法解决在sql中使用什么来代替A not followed B，另一件事是如何将用户I的输出限制在每个orderType和orderCountry最多2个事件，也就是说，如果用户在连续两个窗口中连续5分钟没有在第一个事件之后执行第二个事件，则应该删除该用户的状态，这样我就不会再次获得相同orderType和orderCountry的用户输出。

Answer 1

我认为使用MATCH_RECOGNIZE是不可能的。但是，这可以通过使用DataStream CEP库将超时模式发送到侧输出来实现。

使用KeyedProcessFunction也可以在较低的级别解决这一问题。Apache培训回购中的长途骑行警报演习就是一个例子--如果您愿意，可以直接跳转到解决方案。