KeyValue_KafkaStreams样例

压西妈咪

KeyValue_KafkaStreams样例是一个展示如何使用Kafka Streams API处理键值对数据的示例。该样例演示了如何创建一个简单的Kafka Streams应用程序，用于读取、转换和写入键值对数据到Kafka主题。
KeyValue KafkaStreams 样例

zbhjyg0ijjtzpjz.jpg

（图片来源网络，侵删）
以下是一个使用 Java 编写的简单 KeyValue KafkaStreams 示例，这个示例展示了如何从 Kafka 主题中读取数据，对数据进行处理，并将处理后的数据写入另一个 Kafka 主题。
依赖项
确保你的项目中包含了以下依赖：

   
        org.apache.kafka
        kafkastreams
        2.8.0
   
   
        org.apache.kafka
        kafkaclients
        2.8.0
   
代码示例

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Printed;
import java.util.Properties;
public class KeyValueKafkaStreamsExample {
    public static void main(String[] args) {
        // 配置 Kafka Streams
        Properties config = new Properties();
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "keyvalueexample");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        // 构建 StreamsBuilder
        StreamsBuilder builder = new StreamsBuilder();
        // 从源主题读取数据
        KStream sourceStream = builder.stream("sourcetopic");
        // 对数据进行处理（这里只是简单地将键和值连接起来）
        KStream processedStream = sourceStream.map((key, value) > new KeyValue(key, key + "" + value));
        // 将处理后的数据写入目标主题
        processedStream.to("targettopic", Printed.toSysOut().withLabel("Processed Stream"));
        // 启动 Kafka Streams 应用程序
        KafkaStreams streams = new KafkaStreams(builder.build(), config);
        streams.start();
        // 添加关闭钩子以优雅地关闭应用程序
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}
单元表格

序号	功能描述
1	导入所需的库
2	设置 Kafka Streams 的配置参数
3	创建 StreamsBuilder 实例
4	从源主题读取数据
5	对数据进行处理（这里是一个简单的字符串拼接操作）
6	将处理后的数据写入目标主题
7	启动 Kafka Streams 应用程序
8	添加关闭钩子以确保应用程序在退出时正确关闭

下面是一个关于KeyValue对在Kafka Streams中使用的示例介绍，在这个场景中，假设我们有一个简单的应用程序，它从一个主题接收消息，处理这些消息，并将结果写入另一个主题。

zbhjvsooshpfecc.png

（图片来源网络，侵删）

步骤	描述	Kafka Streams代码样例
1. 创建流处理拓扑	定义一个流处理拓扑，该拓扑会从一个主题接收数据。	KStream stream = builder.stream("input_topic");
2. 处理消息	对接收到的每条消息进行处理，这里使用简单的map函数来演示。	KStream mappedStream = stream.mapValues(value > value.length());
3. 转换键值对	可以选择对键（Key）或值（Value）进行转换，以下示例仅更改了键。	KStream[I] transformedStream = mappedStream.selectKey((key, value) > value);
4. 数据聚合	可以按照键对数据进行聚合，这里假设我们正在对相同的键进行计数。	KTable[I] aggregatedStream = mappedStream.groupByKey().count();
5. 将结果写回主题	将处理后的数据写回到Kafka主题。	aggregatedStream.toStream().to("output_topic", Produced.with(Serdes.Integer(), Serdes.Long()));
6. 启动流处理	启动流处理应用程序。	StreamsConfig config = new StreamsConfig(props); final KafkaStreams streams = new KafkaStreams(builder.build(), config); streams.start();

在这个介绍中，我们定义了一个流处理拓扑，它接收键值对（在本例中，键和值都是字符串类型），并执行以下操作：
使用mapValues对每个值应用一个函数（本例中是计算字符串长度）。
使用selectKey改变键（本例中是根据处理后的值来重新定义键）。
使用groupByKey和count按新键进行聚合。
将处理后的流（一个键和计数值的流）写入新的Kafka主题。
请根据您实际的应用程序需求调整上述代码和步骤，在实际应用中，您需要配置适当的序列化器（Serdes）和Kafka客户端属性。

zbhjk3bx5bqandi.png

（图片来源网络，侵删）