stream kar stream教程 kafka stream和kafka区别

本文档介绍了如何使用 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()；KStreamlt；String， RuleConfiggt；ruleConfigKStream = builder.stream(TOPIC_NAME， Consumed.with(stringSerde，ruleConfigSerde))；登录后复制

接下来，我们根据 SCHEME KStream 使用 groupBy 方法KGroupedStreamlt；String， RuleConfiggt； groupedKStream = RuleConfigKStream.groupBy((key， value) -gt； value.getScheme()， Grouped.with(Serdes.String()， RuleConfigSerde))；登录后复制

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

KTablelt；String， Listlt；RuleConfiggt；gt；ruleStore = groupedKStream.aggregate( ArrayList：：new， (key， value， list) -gt； { list.add(value)； 返回列表； }， Materialized.lt；String， Listlt；RuleConfiggt；， KeyValueStorelt；Bytes， byte[]gt；gt；as(RULE_STORE) .withKeySerde(stringSerde).withValueSerde(listSerde))；登录后复制

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

3. 获取 KTable 中的数据

最后，我们可以使用 kafkaStreams.store 方法获取 KTable 的观点视图，并总结检索数据。ReadOnlyKeyValueStorelt；String， Listlt；RuleConfiggt；gt；ruleKVStore = kafkaStreams.store(StoreQueryParameters.fromNameAndType(RULE_STORE， QueryableStoreTypes.keyValueStore()))；Listlt；RuleConfiggt；ruleConfigs = ruleKVStore.get(quot；MCquot；)；登录后复制

be代码首先通过kafkaStreams.store获取名为RULE_STORE的KTable的视角。然后，使用ruleKVStore.get("MC")检索键为"MC"的RuleConfig对象列表。

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

5. 总结

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

以上就是Kafka Streams：基于按键分组创建全局KTable，值为对象列表的内容，更多请关注乐哥常识网其他相关文章！