如何在FlinkSQL中创建和使用自定义水印生成器？

https://www.80wz.com/zb_users/upload/2025/05/20250529092653174848201355346.txt

将上述代码编译为JAR文件，并上传至Flink集群的lib目录或通过SQL Client加载。

CREATE TABLE my_table (
    event_time TIMESTAMP(3),  -- 事件时间字段
    user_id STRING,
    data STRING
) WITH (
    'connector' = 'kafka',
    'topic' = 'my_topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'watermark' = 'CustomWatermarkStrategy(5000)'  -- 允许5秒延迟
);

SELECT 
    user_id, 
    TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS window_start,
    COUNT(*) AS total_events
FROM my_table
WATERMARK FOR event_time AS CUSTOM_WATERMARK()  -- 使用自定义水印
GROUP BY 
    user_id, 
    TUMBLE(event_time, INTERVAL '1' MINUTE);