kafka流中的聚合和状态存储保持

Question

我有一个像下面这样的用例。对于每个传入的事件，我希望查看某个字段，看看它的状态是否从a变为B，如果是，则将其发送到输出主题。流程是这样的：一个带有键“xyz”的事件以状态A进入，一段时间后另一个带有键“xyz”的事件以状态B进入。

final KStream<String, DomainEvent> inputStream....

final KStream<String, DomainEvent> outputStream = inputStream
          .map((k, v) -> new KeyValue<>(v.getId(), v))
                    .groupByKey(Serialized.with(Serdes.String(), jsonSerde))
                    .aggregate(DomainStatusMonitor::new,
                            (k, v, aggregate) -> {
                                aggregate.updateStatusMonitor(v);
                                return aggregate;
                            }, Materialized.with(Serdes.String(), jsonSerde))
                    .toStream()
                    .filter((k, v) -> v.isStatusChangedFromAtoB())
                    .map((k,v) -> new KeyValue<>(k, v.getDomainEvent()));

有没有更好的方法使用DSL来编写这个逻辑？

上面代码中关于聚合创建的状态存储的两个问题。

null

提前道谢！

Answer 1

>

默认情况下，将使用持久的RocksDB存储。如果要使用内存中的存储，可以传入materialized.as(stores.InMemoryKeyValueStore(...))

如果您有无限数量的唯一键，您最终将耗尽主存或磁盘，您的应用程序将死亡。根据您的语义，您可以通过使用带有大“gap”参数的会话窗口聚合来获得“TTL”，而不是使用过期的旧密钥。

在处理新数据之前，该状态将始终被还原。如果您使用内存存储，则会通过使用基础的changelog主题来实现这一点。根据您所在州的大小，这可能需要一段时间。如果使用持久的RocksDB存储，状态将从磁盘加载，因此不需要还原，处理应该立即发生。只有在松开本地磁盘上的状态时，才会在这种情况下从changelog主题进行还原。