如何使用kafka流保持kafka主题中关键字的N个最新值

Question

假设我使用kafka streams(kafka-streams-scala库，版本2.2.0）。

我想出了一种可能的方法：创建流和可变映射，然后使用stream.foreach来跟踪每个键的N个最近值。

val stream: KStream[String, GenericRecord] = builder.stream[String, GenericRecord]("topicName")

val map = scala.collection.mutable.Map[String, List[MyObject]]

stream.foreach((k, v) =>  {
  //update map
})

我的问题是是否有更好的方法来实现这一点--或者使用streams API，或者至少不使用可变映射。

Answer 1

所以我需要像KTable或GlobalKTable这样的东西，但它们只保留一个值。

继续使用ktable（或globalktable)，但使用结构化值和/或集合作为“值”。Kafka中没有强制要求您将消息值仅限制为基元数据类型（如integer或string)。

想一想:kstream > list。在这里，每个消息都属于一个特定的用户（由userid键标识），并且每个消息都有与该用户关联的0个、1个或多个ClickEvent的列表。这“只是工作”，您只需要为您想要使用的数据类型拥有适当的serdes（序列化器/反序列化器）。

例如，https://github.com/confluentinc/kafka-streams-examples中的CustomStreamTableJoin示例（直接链接到Apache Kafka V2.2的V5.2.1示例）使用pair类在Kafka的消息值中存储一个元组，并且它附带了pairserde。对于存储值的集合也可以这样做（开发人员正在这样做），比如列表 ，正如您在自己的用例中提到的那样。

我需要保留几个最近的key值在kafka主题使用kafka流。[...]我想出了一种可能的方法：创建流和可变映射，[...]

您不需要使用映射。Kafka消息中已经可以使用该键，因此您只需要为消息值提供一个类似列表的数据类型。

或者至少没有可变映射。

您不需要（也不应该）使用可变的数据结构，除非有特定的原因，我认为您的用例中没有这种原因。当正在处理一个新消息并且相应的输出存储在ktable中时，表中为该键存储的任何内容都将被覆盖--因此使用不可变的数据结构作为消息值是完全可以的。