Kafka Streams：基于键分组并创建全局 KTable，值为对象列表

本文档介绍了如何使用 kafka Streams 基于键（例如 Scheme）对数据进行分组，并将结果存储在全局 KTable 中，其中键为字符串（Scheme），值为 RuleConfig 对象列表。通过示例代码演示了如何实现这一目标，并指出了在实现过程中需要注意的关键点，确保 KTable 能够正确地聚合和存储数据。

使用 Kafka Streams 构建 KTable，键为字符串，值为对象列表

在使用 Kafka Streams 处理数据时，经常需要根据某个键对数据进行分组，并将分组后的数据存储起来。本文将介绍如何使用 Kafka Streams 创建一个 KTable，其中键为字符串，值为对象列表。我们将以根据 Scheme 对 RuleConfig 对象进行分组为例，演示如何实现这一目标。

1. 数据准备

假设我们有一个名为 RuleConfig 的类，它包含 SCHEME、RULEORDER 和 Regex 等属性。我们的目标是根据 SCHEME 对 RuleConfig 对象进行分组，并将具有相同 SCHEME 的 RuleConfig 对象存储在一个列表中。

2. Kafka Streams 代码

首先，我们需要创建一个 StreamsBuilder 对象，并从 Kafka 主题中读取数据。

StreamsBuilder builder = new StreamsBuilder(); KStream<String, RuleConfig> ruleConfigKStream = builder.stream(TOPIC_NAME, Consumed.with(stringSerde, ruleConfigserde));

接下来，我们使用 groupBy 方法根据 SCHEME 对 KStream 进行分组。

KGroupedStream<String, RuleConfig> groupedKStream = ruleConfigKStream.groupBy((key, value) -> value.getScheme(), Grouped.with(Serdes.String(), ruleConfigSerde));

然后，我们使用 aggregate 方法将分组后的数据聚合到一个列表中。

KTable<String, List<RuleConfig>> ruleStore = groupedKStream.aggregate(     ArrayList::new,     (key, value, list) -> {         list.add(value);         return list;     },     Materialized.<String, List<RuleConfig>, KeyValueStore<Bytes, byte[]>>as(RULE_STORE)         .withKeySerde(stringSerde).withValueSerde(listSerde) );

在这个代码片段中，ArrayList::new 用于初始化聚合器的初始值（一个空的 ArrayList）。第二个参数是一个 Lambda 表达式，它接受键、值和当前列表作为输入，并将新的 RuleConfig 对象添加到列表中。Materialized 用于指定 KTable 的存储方式，包括存储名称、键序列化器和值序列化器。

3. 获取 KTable 中的数据

最后，我们可以使用 kafkaStreams.store 方法获取 KTable 的只读视图，并从中检索数据。

ReadOnlyKeyValueStore<String, List<RuleConfig>> ruleKVStore = kafkaStreams.store(StoreQueryParameters.fromNameAndType(RULE_STORE, QueryableStoreTypes.keyValueStore()));  List<RuleConfig> ruleConfigs = ruleKVStore.get("MC");

这段代码首先通过 kafkaStreams.store 获取名为 RULE_STORE 的 KTable 的只读视图。然后，使用 ruleKVStore.get(“MC”) 检索键为 “MC” 的 RuleConfig 对象列表。

4. 注意事项

• 确保正确配置了键和值的序列化器和反序列化器。
• Materialized.as 的位置非常重要，它应该作为 aggregate 方法的最后一个参数。
• KTable 中的数据是持久化的，因此需要合理规划存储空间。
• 可以使用 QueryableStoreTypes 提供的不同查询类型来访问 KTable 中的数据。

5. 总结

本文介绍了如何使用 Kafka Streams 创建一个 KTable，其中键为字符串，值为对象列表。通过示例代码演示了如何根据 Scheme 对 RuleConfig 对象进行分组，并将具有相同 Scheme 的 RuleConfig 对象存储在一个列表中。希望本文能够帮助你更好地理解和使用 Kafka Streams。