java - 如何在 Kstream DSL .transform() 方法中访问所有 statestore 数据

Question

尝试访问定义的状态存储中的所有键值，但是在 .transform() 方法中我只能使用一个键(即源键)访问

KeyValueStore<String, String> SS=context.getStateStore("macs");

SS 无法获取 statestore 中的所有键值

SS.get("key1");
SS.get("key2");
SS.get("key3");
SS.get("key4");

4 个返回值中只有 1 个返回值，其余所有返回 null

Answer 1

the SS is not able to get all key values in statestore

这是预期的行为。 Kafka Streams 中“逻辑”状态存储中的数据实际上是跨分布式 Kafka Streams 应用程序的运行实例(即使您只运行 1 个应用程序实例，如 1 个 Docker 容器)的状态存储的实际实例进行分区(分片)为您的应用程序)。下面我来解释一下。

一个简化的例子来说明分区状态存储的性质:如果您的应用程序从具有 5 个分区的输入主题读取，那么该应用程序的处理拓扑将使用 5 个流任务，并且每个流任务将获得一个分区“逻辑”状态存储(参见 Kafka Streams Architecture )。如果您只为您的应用程序运行 1 个应用程序实例(如 1 个 Docker 容器)，那么这个单个实例将执行所有 5 个流任务，但这些流任务是无共享设置——这意味着数据仍然是分区的. Kafka Streams 中的 KTable 也是如此，它们也是以这种方式分区的。

另请参阅:Is Kafka Stream StateStore global over all instances or just local?

您上面的示例仅适用于输入主题只有 1 个分区的特殊情况，因为那时只有 1 个流任务，因此只有 1 个状态存储(可以访问输入数据中的所有可用键).

Trying to access all key values in the defined statestore [...]

现在，如果您确实想要访问输入数据中的所有可用键，您有两个选择(除非您想要沿着只有 1 个分区的输入主题的特殊情况走下去):

• 选项 1:使用全局状态存储(或 GlobalKTable)而不是正常的分区状态存储。可以通过 StreamsBuilder#addGlobalStore(...) 定义/创建全局状态存储，但是 IIRC，您不需要显式地向处理器添加(“附加”)全局存储，而对于正常的状态存储，您必须这样做。相反，任何处理器都可以自动访问全局存储。
• 选项 2:使用 interactive queries feature (也称为可查询状态)在 Kafka Streams 中。

请注意，在这两个选项中，您都可以访问状态存储中的数据仅供阅读。在这两种情况下，您不能直接写入状态存储。如果您需要修改数据，则必须通过用于填充商店的输入主题间接更新它们。