如何在Kafka Streams中对API返回的扁平数据链式调用groupby查询

你之前的思路无法实现需求的核心原因是：你在yearCount步骤已经把原始车辆数据聚合成了「年份-总计数」的KTable，丢失了颜色维度的原始字段，在这之后做任何处理都无法拿到颜色数据，必须从原始的usedCarsInputStream流开始做维度聚合。

实现步骤 #

• 先过滤保留2010年之后的车辆数据，构造「年份+颜色」的复合维度作为分组键
• 分组后统计每个复合维度对应的车辆数
• 再按年份重新聚合，把同一年份下所有颜色的计数汇总为Map结构，最后序列化为你需要的JSON格式即可

示例代码 #

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.kstream.Materialized;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

// 原始输入流保留你已有的定义
KStream<byte[], UsedCars> usedCarsInputStream = 
        builder.stream("used-car-colors", Consumed.with(Serdes.ByteArray(), new UsedCarsSerdes()));

// 1. 过滤2010年之后的车辆数据
KStream<byte[], UsedCars> filteredCars = usedCarsInputStream
        .filter((k, v) -> v.getYear() > 2010);

// 2. 按「年份:颜色」复合键分组统计对应车辆数
KTable<String, Long> yearColorCount = filteredCars
        .groupBy((k, v) -> v.getYear() + ":" + v.getColor())
        .count();

// 3. 按年份重新聚合，把同一年的所有颜色计数汇总成Map
KTable<String, Map<String, String>> yearAggregate = yearColorCount
        .toStream()
        .groupBy((compositeKey, count) -> compositeKey.split(":")[0]) // 提取年份作为分组键
        .aggregate(
                HashMap::new, // 初始化聚合容器
                (year, compositeKey, count, aggMap) -> {
                    String color = compositeKey.split(":")[1].toLowerCase();
                    aggMap.put(color, count.toString());
                    return aggMap;
                },
                Materialized.with(Serdes.String(), Serdes.serdeFrom(HashMap.class))
        );

// 4. 输出符合要求的JSON格式
ObjectMapper objectMapper = new ObjectMapper();
yearAggregate.toStream().foreach((year, colorCountMap) -> {
    try {
        // 包装为你要求的外层结构
        Map<String, Object> output = new HashMap<>();
        output.put(year, Collections.singletonList(colorCountMap));
        // 格式化打印，也可以直接写入下游Kafka主题
        System.out.println(objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(output));
    } catch (JsonProcessingException e) {
        e.printStackTrace();
    }
});

优化建议 #

如果不想用字符串拆分复合键，可以自定义包含year和color字段的POJO作为分组键，同时实现对应的Serde序列化器，性能会比字符串拆分更稳定。如果需要输出所有年份合并的完整大JSON，可以额外加一步全局聚合，把所有年份的结果汇总到同一个Map中输出即可。

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