时间戳

首页 > 解决方案 > 我可以使用 Kafka 流读取和写入不同类型的消息吗？

apache-kafka - 我可以使用 Kafka 流读取和写入不同类型的消息吗？

问题描述

我正在编写一个使用 Kafka 流的应用程序。它从主题 A 读取，进行一些转换，然后写入主题 B。在转换期间，值按键分组，因此输出键、值类型与输入值类型不同。Kafka流使用特定类型的Serdes（例如String serdes对字符串进行序列化和反序列化）进行序列化和反序列化，因此在数据转换后它将不起作用。如何在 Streams API 中定义不同的序列化器和反序列化器？

标签： apache-kafkaapache-kafka-streams

解决方案

你当然可以

当您创建流、调用 groupBy 或将输出写入某个主题时，您可以提供Serde或Serialized. 例子：

Serde<String> stringSerde = Serdes.String();
Consumed<String, String> consumed = Consumed.with(stringSerde, stringSerde);
Produced<String, YourCustomItem> produced = Produced.with(stringSerde, new JsonSerde<>(YourCustomItem.class));

KStream<String, String> kStream = streamsBuilder.stream("sourceTopicName", consumed);
KStream<String, YourCustomItem> transformedKStream = kStream.mapValues((key, value) -> new YourCustomItem());
transformedKStream.to("destinationTopicName", produced);

transformedKStream.groupByKey(Serialized.with(Serdes.String(), new JsonSerde<>(YourCustomItem.class)));

从哪里来JsonSerde的spring-kafka依赖。或者您可以使用以下内容Serde：

Serializer<JsonNode> jsonSerializer = new JsonSerializer();
Deserializer<JsonNode> jsonDeserializer = new JsonDeserializer();
Serde<JsonNode> jsonSerde = Serdes.serdeFrom(jsonSerializer, jsonDeserializer);

推荐阅读