KSQLDB执行建表语句后Kafka主题Key出现多余字符的原因及清理方法咨询

你看到的这些"多余字符"，其实是KSQLDB在窗口聚合（这里是HOPPING窗口）时自动添加的窗口元数据序列化内容。

当你执行带窗口的聚合查询时，KSQLDB默认会把「分组键（也就是你这里的lr.rowkey）+ 窗口的边界信息（比如窗口的起始时间、结束时间）」打包成一个复合结构作为输出消息的Key——这些元数据是KSQLDB内部用来跟踪窗口状态、处理窗口生命周期的必要信息，但对于只需要原始分组键作为业务Key的场景来说，就会显得多余。

要去掉这些窗口元数据，让输出Key只保留你需要的lr.rowkey值，可以通过以下两种方式修改你的KSQLDB语句：

方案1：显式指定输出Key（最直接） #

在CREATE TABLE的查询结尾添加PARTITION BY majmoo（majmoo是你给lr.rowkey起的别名），强制KSQLDB将指定字段作为输出主题的Key，而不是默认的复合窗口Key：

CREATE TABLE LimiteOneMinute_TB2 WITH (kafka_topic='limitoneminute2', value_format='JSON') as 
select 
    (lr.rowkey) majmoo , 
    AS_VALUE(cast(lr.rowkey as varchar)) inet_config, 
    max(lr.inet) ip, 
    (max(lr.rowtime) + 60000) expire_time, 
    true as captcha 
from LIMITER_REQUEST lr 
join CONFIG_TB lc on lr.configid = lc.id 
WINDOW HOPPING (SIZE 59 SECONDS, ADVANCE BY 5 SECONDS) 
group by (lr.rowkey) 
having count(*) > max(lc.oneminute) 
PARTITION BY majmoo  -- 新增此行，指定输出Key为majmoo字段
emit changes;

方案2：配置Key格式参数（适合单字段Key场景） #

如果你的业务Key是单个字段，还可以在WITH子句中添加key_format和wrap_single_value参数，进一步明确Key的序列化规则：

CREATE TABLE LimiteOneMinute_TB2 WITH (
    kafka_topic='limitoneminute2', 
    value_format='JSON',
    key_format='JSON',  -- 指定Key使用JSON格式序列化
    wrap_single_value=true  -- 当Key为单个字段时，不使用复合结构包裹
) as 
select 
    (lr.rowkey) majmoo , 
    AS_VALUE(cast(lr.rowkey as varchar)) inet_config, 
    max(lr.inet) ip, 
    (max(lr.rowtime) + 60000) expire_time, 
    true as captcha 
from LIMITER_REQUEST lr 
join CONFIG_TB lc on lr.configid = lc.id 
WINDOW HOPPING (SIZE 59 SECONDS, ADVANCE BY 5 SECONDS) 
group by (lr.rowkey) 
having count(*) > max(lc.oneminute) 
PARTITION BY majmoo 
emit changes;

修改后重新执行语句，你可以用KSQLDB的PRINT命令查看输出主题的内容：

PRINT limitoneminute2 FROM BEGINNING;

此时输出消息的Key应该只会包含你原始的lr.rowkey值，窗口相关的多余字符已经被清理掉了。

内容的提问来源于stack exchange，提问作者Alihossein shahabi