Kafka Streams按person name聚合为KTable遇阻，求解决方案

你的核心问题是当前处理后的流key为null，Kafka Streams的分组操作依赖key区分不同分组，导致无法按person name正确聚合。需要先修正key，再执行分组聚合，具体步骤如下：

1. 修正流的键值对：将key设置为person的name #

修改流处理逻辑，把原始数据中的person.name提取为新的key，同时保留person的JSON字符串作为value：

KStream<String, String> personStream = builder.stream("topic-1")
    .map((originalKey, value) -> {
        JSONObject json = new JSONObject(value);
        JSONObject personJson = json.getJSONObject("person");
        // 提取person的name作为分组的key
        String personName = personJson.getString("name");
        // 返回新的键值对
        return KeyValue.pair(personName, personJson.toString());
    });

2. 分组并聚合为KTable #

通过groupByKey按name分组，再用aggregate操作将同name的person JSON收集到列表中，需配置正确的序列化器处理列表类型：

第一步：配置列表的JSON序列化器 #

// 用于序列化/反序列化List<String>的Serde
Serde<List<String>> listSerde = Serdes.serdeFrom(
    new JsonSerializer<>(),
    new JsonDeserializer<>(List.class)
);

第二步：执行分组聚合 #

KTable<String, List<String>> groupedPersonTable = personStream
    .groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
    .aggregate(
        // 初始化聚合状态：创建空列表
        ArrayList::new,
        // 聚合逻辑：将新的person JSON添加到列表
        (personName, newPersonJson, currentList) -> {
            currentList.add(newPersonJson);
            return currentList;
        },
        // 合并逻辑：分布式场景下合并两个分区的列表
        (personName, listA, listB) -> {
            listA.addAll(listB);
            return listA;
        },
        // 指定状态存储的Serde和名称
        Materialized.<String, List<String>, KeyValueStore<Bytes, byte[]>>as("grouped-persons-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(listSerde)
    );

关键说明 #

• 之前参考的方案无效，是因为你的流key为null，所有数据会被分到同一个分组，无法区分不同name的person。必须先将key设置为分组字段（person.name），分组操作才能生效。
• 如果需要对同name的person去重（比如保留每个person id的最新数据），可以将聚合状态改为Map<String, String>（key为person.id，value为最新的JSON），最后再转换为列表。
• 确保序列化器配置正确，否则状态存储会因无法序列化列表数据抛出异常。

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