问题标签 [ktable]

我有大量的 IP 地址前缀和相关数据，我想将它们加入到另一个包含 IP 地址的流中。

问题是这不是一个简单的地图/查找。通常，IP 前缀应存储在 trie（如自适应基数 trie）中，其中可以使用 IP 地址作为输入来执行最长前缀匹配。

有没有办法用卡夫卡做到这一点？可以实现更复杂的数据存储和查找吗？

目前，我们在 Kafka Streams 应用程序中使用以下内容：

我刚刚意识到，加入后的 mapValues 创建了一个额外的状态存储。

如果 valueMapper 中的计算在某种程度上是微不足道的（例如删除对象中的字段等），那么避免额外的 statestore 不是更好吗？我是否需要转换为 KStream 并使用 KStream.mapValues 来避免 stateStore，即

还是有更好的选择来在加入后应用附加映射？

我们有 2 个微服务：

1. 负责消费消息并使用配置信息更新 globalKtable (key=id, value=myObject)

2. 一旦使用某个 id 作为输入激活它，它应该查看配置 globalKtable 并通过该 id 检索数据

如何从 app2 访问该 globalktable。是否应该添加 REST 层来检索信息或 Stream 具有其他一些功能？

谢谢

KStream 和 KTable 可以一起使用吗？我正在尝试实现 KafkaReceiver，因为它认为它会处理背压，但它会变成一个 KTable。

谁能帮忙。我有以下要求。

要求：处理非重复，订购聊天消息并根据 ProgramUserId 将它们捆绑在一起，这是过程和涉及的主题。

数据设置：ProgramUserId 可以有任意数量的消息，但每条消息都是唯一的，并且有一个复合键：MsgId + Action。所以想象一下kafka中的数据如下所示。

P2->M3+A1 , P2->M2+A1 , P2->M1+A1 , P1->M3+A1 , P1->M2+A2 , P1->M2+A1 , P1->M1+A1

我现在正在这样做：

Initial-Topic：（原始密钥：ProgramUserId）

1）从 Initial-Topic --> 使用 Kstream（重新键入：Msg Id + Action）--> 然后写入主题：dedup-Topic

2. 从dedup-Topic --> 使用 Kstream（重新键入原始密钥：ProgramUserId）--> 写入主题： Final-Topic

由于我们在 dedup-topic 上重新键入，因此消息的顺序会混乱，因为重新键入会导致重新分区，因此订单中没有保证。

我添加了以下逻辑来实现重复数据删除：从dedup-topic创建 Ktable 和 Postgres 表（使用 Sink 连接）。对于每条传入的消息，检查 Ktable 和 PG 表中的键（Msg Id + Action）。如果未找到记录，则表示它没有重复并将该记录写入 dedup-topic。

但是由于在dedup-Topic 中重新键入/重新分区，上述消息顺序混乱。

请帮助此时如何实现有序的消息？

是否可以交叉引用表中字段内的部分，如下所示？

当我尝试上面的代码时，它说

错误：“@”是开头“@”的字符串中无法识别的转义

关于KStream - KTable连接的问题。通常这种连接用于 KTable 提供参考数据的数据丰富目的。

所以问题是，当 KTable 记录得到更新时，我们如何去更新我们已经处理、丰富并可能存储在某个数据存储中的旧记录？

有没有我们可以遵循的模式？

（请假设KTable - KTable不会是一个选项，因为KStreamside 会发出大量的变化）