java - 卡夫卡流 : how to convert data types?

Question

我正在使用 Kafka 的 Streams API 和拓扑构建器。我想知道如何才能拥有一个可以将一种数据类型转换为另一种数据类型的处理器，以便管道中的下一个处理器可以使用它。

作为一个简单的用例:

[topic]--(string)-->[processor: parse json]--(object)-->[processor 2]--(object)-->[sink]

有什么想法吗？

Answer 1

我假设您想将 Kafka 主题中的消息值从字符串转换为 JSON。

你只需要两部分:

• 将字符串转换为 JSON(或 Pojo)的代码。在这里选择你需要的任何东西，例如有几个 Java 库可以让这一切变得简单。
• 在 Kafka Streams 中，定义 (1) String 的值 serde 用于从源主题读取，以及 (2) 定义相应的值 serde 用于将 JSON 数据(或 Pojo)写入目的地主题。 Serdes 需要在需要的时间/地点具体化您的数据(例如，将您的 Pojo 写入 Kafka 需要具体化)。

请参阅 https://github.com/apache/kafka/tree/trunk/streams/examples/src/main/java/org/apache/kafka/streams/examples/pageview 下的示例代码如何将 JSON 与 Apache Kafka 的 Streams API 结合使用。