基于Kafka Streams的渐进式键约束聚合实现可行性咨询

你的需求完全可以通过Kafka Streams的自定义聚合器结合状态存储来实现，核心是在两级聚合中分别维护必要的中间状态，确保tombstone事件能触发正确的结果更新。以下是具体的实现思路和代码示例：

核心需求拆解 #

1. 第一级聚合：将A_B_C格式的键截断为A_B，分组后维护该分组下所有子键（C部分）的数值，动态计算当前最大值；收到tombstone（值为null）时移除对应子键并重新计算最大值。
2. 第二级聚合：将A_B格式的键截断为A，分组后维护所有A_*分组的当前最大值，生成去重的集合；当某个A_*的最大值变化或被删除时，同步更新集合内容。

分步实现 #

1. 第一级聚合：按A_B分组取最大值（支持Tombstone） #

这里不能直接用内置的max()聚合器，因为内置聚合器无法跟踪分组内的所有子键——当收到tombstone时，需要移除对应子键后重新计算最大值，而非直接删除整个分组的状态。

// 假设原始流为KStream<String, Integer> originalStream
KGroupedTable<String, Integer> groupedByAB = originalStream
    .groupBy(
        (fullKey, value) -> fullKey.split("_", 2)[0], // 截断键为A_B
        Grouped.with(Serdes.String(), Serdes.Integer())
    );

// 自定义聚合器：维护A_B分组下所有子键(C)的数值映射
Table<String, Integer> abMaxTable = groupedByAB
    .aggregate(
        // 初始化状态：空HashMap存储子键到值的映射
        HashMap::new,
        // 聚合逻辑：处理新增/删除事件，计算当前最大值
        (abKey, fullKey, value, subKeyMap) -> {
            String subKey = fullKey.split("_")[2]; // 提取子键C部分
            if (value == null) {
                // Tombstone事件：移除对应子键
                subKeyMap.remove(subKey);
            } else {
                // 更新子键对应的数值
                subKeyMap.put(subKey, value);
            }
            // 计算当前最大值，状态为空则返回null（输出tombstone）
            return subKeyMap.isEmpty() ? null : Collections.max(subKeyMap.values());
        },
        // 配置状态存储：指定存储名称和序列化方式
        Materialized.<String, Map<String, Integer>, KeyValueStore<Bytes, byte[]>>as("ab-subkey-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.serdeFrom(new JsonSerde<>(), new JsonSerde<>()))
    );

2. 第二级聚合：按A分组生成去重最大值集合 #

同样需要自定义聚合器，跟踪每个A分组下所有A_*的当前最大值，确保tombstone事件触发时能正确更新去重集合。

首先定义状态对象（用于存储中间数据）：

// 自定义状态对象：跟踪A_*分组的最大值，以及去重后的集合
class AGState {
    // 存储A_B -> 当前最大值的映射
    public Map<String, Integer> abToMax = new HashMap<>();
    // 存储去重后的最大值集合
    public Set<Integer> uniqueMaxValues = new HashSet<>();
}

然后实现聚合逻辑：

KGroupedTable<String, Integer> groupedByA = abMaxTable
    .toStream()
    .groupBy(
        (abKey, maxValue) -> abKey.split("_")[0], // 截断键为A
        Grouped.with(Serdes.String(), Serdes.Integer())
    );

Table<String, Set<Integer>> aUniqueMaxTable = groupedByA
    .aggregate(
        // 初始化状态
        AGState::new,
        // 聚合逻辑：更新A分组的状态集合
        (aKey, abKey, newMaxValue, state) -> {
            Integer oldMaxValue = state.abToMax.get(abKey);
            
            // 处理旧值：如果该值没有其他A_*分组持有，从去重集合中移除
            if (oldMaxValue != null) {
                long count = state.abToMax.values().stream()
                    .filter(val -> val.equals(oldMaxValue))
                    .count();
                if (count <= 1) {
                    state.uniqueMaxValues.remove(oldMaxValue);
                }
            }
            
            // 处理新值
            if (newMaxValue == null) {
                // Tombstone事件：移除A_B的映射
                state.abToMax.remove(abKey);
            } else {
                // 更新A_B的最大值映射
                state.abToMax.put(abKey, newMaxValue);
                // 将新值加入去重集合
                state.uniqueMaxValues.add(newMaxValue);
            }
            
            return state;
        },
        // 配置状态存储
        Materialized.<String, AGState, KeyValueStore<Bytes, byte[]>>as("a-unique-max-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.serdeFrom(new JsonSerde<>(AGState.class), new JsonSerde<>(AGState.class)))
    )
    // 提取去重集合作为最终输出值
    .mapValues(state -> state.uniqueMaxValues);

关键注意事项 #

• 状态序列化：自定义状态对象（如HashMap、AGState）需要实现序列化逻辑，示例中使用JSON序列化，你也可以选择Avro等更高效的序列化方式。
• Tombstone传递：当第一级聚合的状态为空（所有子键被删除）时，要输出tombstone（null），确保第二级聚合能正确处理分组删除事件。
• 性能优化：如果数据量较大，可配置状态存储的清理策略（如过期时间），或使用 RocksDB 作为状态存储引擎提升性能。
• 异常处理：建议添加键格式校验逻辑，过滤不符合A_B_C格式的记录，避免聚合过程中出现数组越界等异常。

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