如何基于Kafka状态存储中值的字段获取键列表

我已定义如下Kafka状态存储：

StoreBuilder<KeyValueStore<String, DataDocument>> indexStore = Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("Data"), Serdes.String(), DataDocumentSerDes.dataDocumentSerDes())
            .withLoggingEnabled(changelogConfig);  

同时已实现处理器，将源主题的数据写入该状态存储。我的需求是基于DataDocument的某个字段值，获取对应的主键列表。

我目前想到两种方案，但都有明显缺陷：

1. 遍历全量键值对过滤
   直接获取状态存储的所有键值对，逐个判断字段是否匹配，然后收集符合条件的主键。但这种方式需要遍历全量数据，在数据量大时开销极高：

   ReadOnlyKeyValueStore<String, DataDocument> roStore = streams.store(StoreQueryParameters.fromNameAndType("Data"),
            QueryableStoreTypes.<String, DataDocument>keyValueStore());
    KeyValueIterator<String, DataDocument> kvItr = roStore.all();
    while(kvItr.hasNext()) {
        if(kvItr.next().value.isField()) {
            //Store to a list
        }
    }
   

2. 为每个查询字段单独建状态存储
   针对每个需要查询的字段，创建独立的状态存储，以字段值为键，主键列表为值。但这种方式扩展性极差，新增查询字段就需要修改拓扑、新增存储，维护成本很高。

请问有没有基于Kafka拓扑的更优实现方案？

可以通过维护一个通用反向索引存储来解决这个问题，既避免全量遍历，又具备扩展性，不需要为每个字段单独建存储。

实现思路 #

创建一个额外的KeyValueStore，键的格式为{字段名}:{字段值}，值为对应主键的集合（比如用Set<String>存储）。在主存储的处理器逻辑中，同步维护这个索引：

• 当写入/更新主存储的DataDocument时，先处理旧值（如果存在）：从索引中移除旧字段值对应的主键；
• 再处理新值：将主键添加到新字段值对应的索引条目里；
• 对于不需要索引的字段，可以跳过处理。

代码示例 #

1. 定义索引存储 #

// 索引存储：键为"fieldName:fieldValue"，值为主键集合
StoreBuilder<KeyValueStore<String, Set<String>>> inverseIndexStore = Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("Data-Inverse-Index"),
        Serdes.String(),
        // 自定义Serde序列化Set<String>，也可以用JSON Serde替代
        Serdes.serdeFrom(new StringSerializer(), new StringDeserializer()) {
            @Override
            public Serializer<Set<String>> serializer() {
                return (topic, data) -> data != null ? String.join(",", data).getBytes() : null;
            }

            @Override
            public Deserializer<Set<String>> deserializer() {
                return (topic, data) -> data != null ? new HashSet<>(Arrays.asList(new String(data).split(","))) : new HashSet<>();
            }
        }
).withLoggingEnabled(changelogConfig);

2. 在处理器中同步维护索引 #

在写入主存储的处理器里，添加索引维护逻辑：

processorContext.addProcessor("data-processor", () -> new Processor<String, DataDocument>() {
    private KeyValueStore<String, DataDocument> mainStore;
    private KeyValueStore<String, Set<String>> inverseIndex;

    @Override
    public void init(ProcessorContext context) {
        mainStore = context.getStateStore("Data");
        inverseIndex = context.getStateStore("Data-Inverse-Index");
    }

    @Override
    public void process(String key, DataDocument newDoc) {
        // 1. 获取旧数据，处理旧字段的索引移除
        DataDocument oldDoc = mainStore.get(key);
        if (oldDoc != null) {
            // 假设我们要索引的字段是"targetField"，可根据需求扩展为多个字段
            String oldFieldKey = "targetField:" + oldDoc.getTargetField();
            Set<String> oldKeys = inverseIndex.get(oldFieldKey);
            if (oldKeys != null) {
                oldKeys.remove(key);
                if (oldKeys.isEmpty()) {
                    inverseIndex.delete(oldFieldKey);
                } else {
                    inverseIndex.put(oldFieldKey, oldKeys);
                }
            }
        }

        // 2. 写入主存储
        mainStore.put(key, newDoc);

        // 3. 处理新字段的索引添加
        String newFieldKey = "targetField:" + newDoc.getTargetField();
        Set<String> newKeys = inverseIndex.get(newFieldKey);
        if (newKeys == null) {
            newKeys = new HashSet<>();
        }
        newKeys.add(key);
        inverseIndex.put(newFieldKey, newKeys);
    }

    @Override
    public void close() {}
}, "source-topic-processor");

// 将索引存储添加到拓扑
topology.addStateStore(inverseIndexStore, "data-processor");

3. 查询索引获取主键列表 #

查询时直接根据字段名:字段值获取对应的主键集合，无需全量遍历：

ReadOnlyKeyValueStore<String, Set<String>> roIndexStore = streams.store(
        StoreQueryParameters.fromNameAndType("Data-Inverse-Index"),
        QueryableStoreTypes.keyValueStore()
);
// 查询targetField值为"xxx"的所有主键
Set<String> keys = roIndexStore.get("targetField:xxx");

扩展性说明 #

如果需要新增查询字段，只需要在处理器的索引维护逻辑中，添加对应字段的处理代码即可，不需要新增状态存储。同时可以通过配置或动态配置的方式，指定需要索引的字段，进一步提升扩展性。

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