如何在 kafka 主题中仅存储最新的键值答案

【问题标题】：How to store only latest key values in a kafka topic如何在 kafka 主题中仅存储最新的键值
【发布时间】：2019-07-05 13:08:05
【问题描述】：

我有一个有数据流的主题。我需要的是从该主题创建一个单独的主题，该主题仅具有给定键的最新值集。

我认为 KTable 的全部目的是存储给定键的最新值，而不是存储整个事件流。但是我似乎无法让它工作。运行下面的代码会生成密钥库，但该密钥库（maintopiclatest）中有一个事件流（不仅仅是最新值）。因此，如果我两次发送主题中包含 1000 条记录的请求，而不是看到 1000 条记录，我看到的是 2000 条记录。

var serializer = new KafkaSpecificRecordSerializer();
var deserializer = new KafkaSpecificRecordDeserializer();

var stream = kStreamBuilder.stream("maintopic",
    Consumed.with(Serdes.String(), Serdes.serdeFrom(serializer, deserializer)));

var table = stream
    .groupByKey()
    .reduce((aggV, newV) -> newV, Materialized.as("maintopiclatest"));

另一个问题是，如果我想将 KTable 存储在一个新主题中，我不知道该怎么做。为了做到这一点，我似乎必须将它转回 Stream 以便我可以在其上调用“.to”。但是，其中包含整个事件流，而不仅仅是最新值。

【问题讨论】：

为什么不用日志压缩主题？
@Mukeshprajapati 这是我尝试的第一件事，但似乎 Log Compaction 也没有像宣传的那样工作。无论我如何调整与之相关的各种配置设置，它似乎都不会立即启动。而且我读到它只有在存在一定数量的数据后才会在特定条件下启动。然后我阅读了有关 KTables 的信息，这让我觉得它们正是为这种只需要给定键的最新值的场景而制作的。然而，它具有所有的历史价值。我一定是错过了什么。

标签： scala apache-kafka apache-kafka-streams spring-kafka

【解决方案1】：

这不是 KTable 的工作方式。

KTable 本身有一个内部状态存储，每个键只存储一条记录。但是，KTable 会不断更新并受制于所谓的 stream-table-duality。 KTable 的每次更新都会作为变更日志记录发送到下游：https://docs.confluent.io/current/streams/concepts.html#duality-of-streams-and-tables。因此，每个输入记录都会产生一个输出记录。

因为是流处理，所以没有“每个值的最后一个键”。

我有一个有数据流的主题。我需要的是从该主题创建一个单独的主题，该主题仅具有给定键的最新值集。

您希望 KTable 在哪个时间点发出更新？这个问题没有答案，因为输入流在概念上是无限的。

【讨论】：

“我认为 KTable 的整个目的类似于数据库表，仅按键存储‘最新’值”——这是正确的，但您必须将其视为“快照” “随着时间的推移而演变。 “我很困惑”——这种情况经常发生——这是一种完全不同的方法，你需要建立一个新的心理模型（这需要时间......）——你还必须区分当前的 KTable 状态和它的变更日志主题——当前状态每个键只存储一条记录——但是，变更日志主题存储一段时间内的更新。
只有在变更日志主题压缩后，旧值才会被删除，并且每个键只保留最新的记录。请注意，如果您处理 KTable，您不会处理快照，而是处理变更日志主题。 -- KTables 是物化的客户端（默认使用 RocksDB 键值存储）。如果要“时间点”查找，进入当前KTable状态，可以使用“交互式查询”：docs.confluent.io/current/streams/developer-guide/…
“当前状态”表示“当前处理状态”——如果你使用builder.table("topic")，默认情况下处理从最早的偏移量开始。从而消费了完整的topic，处理过程中更新更新KTable状态。假设上游生产者应用程序不断写入"topic"，处理永远不会完成（注意，这是正常情况，因为它是流处理）。因此，在处理“当前状态”的过程中，表示从“开始到当前偏移”的数据对应的状态（当前偏移
因此，不清楚您所说的“当前状态”是什么意思——如果您假设没有上游生产者正在写入"topic"，则 KTable 将不会在结束后更新已到达主题（当然，应用程序仍在运行，等待对主题的新写入，它将立即用于更新 KTable）。在您使用 IQ 查询状态的任何时间点，它都会返回存储中的任何内容（取决于 atm 处理的时间点）。如果没有发生更新，您显然会得到“最新状态”。
正确。对于toStream()，它会为您提供完整的历史记录（注意，当您到达输入主题的末尾时，toStream() 不会开始处理，但它在处理输入主题的同时运行——它基本上创建处理时所有表更新的变更日志流）。如果您想迭代存储中的所有键（对于某些“快照”），您可以通过store.all() 使用 IQ。