Kafka窗口流问题：窗口后Key异常及历史数据清理需求

问题拆解 #

你现在遇到两个核心问题：

1. 窗口化后Key出现乱码（比如2023010612����），因为窗口化后默认输出的是带窗口时间信息的WindowedKey，直接序列化就会多出乱码字符
2. 需要通过窗口化自动清理两天前的数据，同时保留原始Key格式

另外你的aggregate逻辑其实没实现“统计日数据量”的需求——现在是直接用新value覆盖聚合结果，相当于只保留最后一条数据，不是累加计数。

修正后的代码 #

// 先过滤掉两天前的输入数据，避免无效计算
stream.filter((key, value) -> {
    // 从value里解析出时间Key（格式yyyyMMddHH）
    JsonNode json = new ObjectMapper().readTree(value);
    String timeKeyStr = json.get("Key").asText();
    LocalDateTime eventTime = LocalDateTime.parse(timeKeyStr, DateTimeFormatter.ofPattern("yyyyMMddHH"));
    // 只保留两天内的事件
    return eventTime.isAfter(LocalDateTime.now().minusDays(2));
})
.groupByKey()
// 配置24小时滚动窗口，5分钟延迟容忍期，窗口保留24小时（过期自动清理）
.windowedBy(TimeWindows.of(Duration.ofHours(24))
        .grace(Duration.ofMinutes(5))
        .withRetention(Duration.ofHours(24)))
// 修正聚合逻辑：累加Count值，实现日数据量统计
.aggregate(
        () -> 0, // 初始计数为0
        (key, value, totalCount) -> {
            JsonNode json = new ObjectMapper().readTree(value);
            return totalCount + Integer.parseInt(json.get("Count").asText());
        },
        // 指定状态存储，同时设置状态保留2天，清理旧数据
        Materialized.<String, Integer, KeyValueStore<Bytes, byte[]>>as("daily-count-store")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.Integer())
                .withRetention(Duration.ofDays(2))
)
// 直到窗口关闭才输出最终结果，避免中间数据
.suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
// 从WindowedKey里提取原始Key，恢复成窗口前的格式
.toStream(windowedKey -> windowedKey.key())
.to("Test_windowing1");

关键修改说明 #

• 恢复原始Key：toStream(windowedKey -> windowedKey.key())是核心——窗口化后生成的WindowedKey包含原始Key和窗口时间区间，这里直接提取原始字符串Key，就不会出现乱码了。
• 清理历史数据：
  • 输入阶段加filter，直接过滤掉两天前的事件，不让旧数据进入窗口计算
  • 窗口配置withRetention(Duration.ofHours(24))，窗口关闭后自动删除窗口数据
  • 状态存储设置withRetention(Duration.ofDays(2))，确保状态里不会留超过两天的旧数据
• 修正统计逻辑：把原来的覆盖逻辑改成累加Count值，真正实现“统计过去24小时数据量”的需求

额外提醒 #

一定要确保你的Kafka Streams应用是基于事件时间处理的，建议配置自定义时间戳提取器，从value的Key字段解析时间戳，不然窗口会用处理时间计算，统计结果会不准。如果不需要中间结果，suppress可以减少下游主题的数据量，只输出窗口最终统计值。

内容的提问来源于stack exchange，提问作者Darshan Shirke