如何在光闪sql表中自动生成水印？

Question

我正在测试flink-cep-sql，我的水印定义为行时间，我的表是kafka表。由于水印依赖于所有kafka分区中的最小值，所以每个新消息都必须等待kafka划分对齐，然后cep触发结果。

我的kafka表(主题有3个分区)定义为

create table test_table(
    agent_id String, room_id String, 
    create_time Bigint, 
    call_type String, 
    application_id String, 
    connect_time Bigint, 
    row_time as to_timestamp_ltz(create_time, 3), 
    WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
)

这是我的cep sql

select * from test_table  match_recognize (
   partition by agent_id,room_id,call_type 
   order by row_time
   measures  
       last(BF.create_time) as create_time, 
       last(AF.connect_time) as connect_time 
   one row per match after match SKIP PAST LAST ROW 
   pattern (BF+ AF) WITHIN INTERVAL '1' HOUR 
   define 
       BF as BF.connect_time = 0,
       AF as AF.connect_time > 0 and BF.room_id = AF.room_id and BF.call_type = AF.call_type 
) as T ;

cep sql触发结果是正确的，但总是迟到，因为每个分区都需要对齐水印。如何立即获得最新结果或在flink sql表中自动生成水印？

Answer 1

您的模式要求查找< code>connect_time的行

如果您愿意放松模式匹配语义学，为什么不将此MATCH_RECOGNIZE查询替换为间隔连接（带有时间约束的自连接）。有关详细信息，请参阅https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/queries/joins/#interval-joins。

顺便说一下，这部分定义了AF

... and BF.room_id = AF.room_id and BF.call_type = AF.call_type

没有任何影响，因为流已经由room_id和call_type分区。