flink cep sql事件未触发

Question

我在Flink SQL中使用了CEP模式，它按照预期连接到Kafka broker。但是当我连接到基于集群的云kafka设置时，Flink CEP没有触发。以下是我的sql:

create table agent_action_detail 
(
    agent_id String, 
    room_id String, 
    create_time Bigint, 
    call_type String, 
    application_id String, 
    connect_time Bigint, 
    row_time TIMESTAMP_LTZ(3), WATERMARK for row_time as row_time  - INTERVAL '1' MINUTE) 
with ('connector'='kafka', 'topic'='agent-action-detail', ...)

然后我以json格式发送消息，如

{"agent_id":"agent_221","room_id":"room1","create_time":1635206828877,"call_type":"inbound","application_id":"app1","connect_time":1635206501735,"row_time":"2021-10-25 16:07:09.019Z"}

在 flink Web ui 中，水印工作精细 flink Web ui

我运行我的cep sql：

select * from agent_action_detail
 match_recognize(
    partition by agent_id 
    order by row_time 
    measures 
        last(BF.create_time) as create_time, 
        first(AF.connect_time) as connect_time 
    one row per match AFTER MATCH SKIP PAST LAST ROW 
    pattern (BF+ AF) define BF as BF.connect_time > 0 ,AF as AF.connect_time > 0
 )

每个Kafka消息，connect_

select * from agent_action_detail match_recognize( partition by agent_id order by row_time  measures AF.connect_time as connect_time one row per match pattern (BF AF) WITHIN INTERVAL '1' second define BF as (last(BF.connect_time, 1) < 1), AF as AF.connect_time >= 100)

这是另一个仍然不起作用的cep sql。并且agent_action_detail表被另一个flink sql作为

insert into agent_action_detail select data.agent_id, data.room_id, data.create_time, data.call_type, data.application_id, data.connect_time, now() from source_table where type = 'xxx'

Answer 1

有几种情况可能导致模式匹配不产生任何结果：

• 输入实际上不包含模式
• 水印不正确
• 该模式在某种程度上是病态的

这个特殊的模式没有退出条件。这种模式不允许模式匹配引擎的内部状态被清除，这将导致问题。

如果您直接使用Flink CEP，我会告诉您尝试添加直到（条件）或在（时间）内，以限制可能匹配的数量。

使用MATCH_RECOGNIZE，查看是否可以向模式添加不同的终止元素。

更新：由于您在修改模式后仍然没有得到任何结果，因此您应该确定水印是否是问题的根源。CEP依赖于按时间对输入流进行排序，这取决于水印——但前提是您使用的是事件时间。

最简单的测试方法是切换到使用流转时长：

create table agent_action_detail 
(
    agent_id String, 
    ...
    row_time AS PROCTIME()
)
with (...)

如果这行得通，那么时间戳或水印就是问题所在。例如，如果所有事件都延迟了，您将得不到任何结果。对于您的情况，我想知道row_time列中有没有数据。

如果这不能揭示问题，请分享一个最小的可重复的例子，包括观察问题所需的数据。