Kafka流内部数据管理

Question

在我的公司，我们广泛使用Kafka，但出于容错的原因，我们一直使用关系数据库来存储几个中间转换和聚合的结果。现在我们正在探索Kafka流作为一种更自然的方法来做到这一点。通常，我们的需求很简单--其中一个例子是

• 监听 、 、 、 ...
的输入队列
• 对每个记录执行一些高延迟操作（调用远程服务）
• 如果在处理 时， 都已生成，那么我应该处理V3，因为V2已经过时了

为了实现这一点，我将主题作为ktable阅读。代码如下所示

KStreamBuilder builder = new KStreamBuilder();
KTable<String, String> kTable = builder.table("input-topic");
kTable.toStream().foreach((K,V) -> client.post(V));
return builder;

这是预期的，但我不清楚Kafka是如何自动实现这一点的。我假设Kafka创造了内部主题来实现这一点，但我没有看到任何内部主题被创造出来。Javadoc的方法似乎证实了这一点。但后来我看到了这个官方页面，它似乎表明Kafka使用了一个单独的数据存储库，也就是RocksDB，以及一个changelog主题。

现在我很困惑，因为changelog主题是在什么情况下创建的。我的问题是

null

Answer 1

>

Kafka Streams存储本地状态。默认情况下使用rocksdb。然而，局部状态是短暂的。为了容错，对存储区的所有更新也被写入changelog主题中。这允许在出现故障或伸缩/扩展时重建和/或迁移存储区。对于您的特殊情况，没有创建changelog主题，因为ktable不是聚合的结果，而是直接从主题填充的--这只是一个优化。因为changelog主题将包含与输入主题完全相同的数据，所以Kafka Streams避免了数据重复，并在出现错误时使用输入主题作为changelog主题。

我不知道你这个问题到底是什么意思。注意，RocksDB被认为是一个短暂的存储。默认情况下使用RocksDB，原因如下：为什么Apache Kafka Streams使用RocksDB，以及如何更改它？（例如，它允许保存大于主存的状态，因为它可能溢出到磁盘）。您可以用任何其他存储替换RocksDB。Kafka Streams还附带内存存储。（编辑）

>

这是正确的。您需要相应地提供应用程序，以便能够存储整体状态的局部碎片。这里有一个大小调整指南：https://docs.confluent.io/current/streams/sizing.html

1. 如果启用日志压缩，则忽略保留时间设置。因此，是的，最新的更新将永远保持（或者直到写入value=null的墓碑消息）。注意，当在代理端执行压缩时，旧数据将被垃圾收集，因此，在还原时，只读取每个键的新版本--旧版本在压缩过程中被删除。如果您在一段时间后对某些数据不感兴趣，则需要在源主题中写入一个墓碑以使其工作。