2

我正在使用 Kafka Streams 0.10.1.1 版本。

状态存储的 RocksDB 实现无法处理我们的 50k/msg 速率，所以我想将状态存储更改为内存中的存储。根据文档，这应该是可能的：http: //docs.confluent.io/3.1.0/streams/architecture.html#state

但是，当我实现这个时：

val stateStore = Stores.create(stateStoreName).withStringKeys().withStringKeys().inMemory().build()

val procSuppl: KStreamAggregate = ... // I'll spare the implementation details

streamBuilder.addSource(
  "mysource",
  new StringDeserializer(),
  new StringDeserializer(),
  "input_topic"
).addProcessor("proc", procSuppl,  "mysource").addStateStore(stateStore, "proc")

我最终在运行时出现此错误：

Caused by: java.lang.ClassCastException: org.apache.kafka.streams.state.internals.MeteredKeyValueStore cannot be cast to org.apache.kafka.streams.state.internals.CachedStateStore
2017-01-23T13:19:11.830674020Z  at org.apache.kafka.streams.kstream.internals.KStreamAggregate$KStreamAggregateProcessor.init(KStreamAggregate.java:62)

上述方法的实现是：

public void init(ProcessorContext context) {
        super.init(context);
        store = (KeyValueStore<K, T>) context.getStateStore(storeName);
        ((CachedStateStore) store).setFlushListener(new ForwardingCacheFlushListener<K, V>(context, sendOldValues));
    }

为什么它试图将状态存储转换为CachedStateStore实例？我怎样才能实现一个简单的内存状态存储，根据文档应该是可能的？

谢谢

4

1 回答 1

6

为了创建一个内存状态存储，需要创建一个存储供应商（使用Stores工厂对象）：

val storeSupplier = Stores.inMemoryKeyValueStore("in-mem")

那么在实现 KTable 时需要使用 store 供应商：

val wordCounts =  builder
  .stream[String, String]("streams-plaintext-input")
  .flatMapValues(textLine => textLine.toLowerCase.split("\\W+"))
  .groupBy((_, word) => word)
  .count()(Materialized.as(storeSupplier))

获取可查询存储：

val qStore = streams.store(
  wordCounts.queryableStoreName,
  QueryableStoreTypes.keyValueStore[String, Long])
于 2018-10-25T14:27:48.713 回答