Как получить отсортированный KeyValueStore из KTable?

Question

Я хочу материализовать KTable из KStream и хочу, чтобы KeyValueStore сортировался по Key.

Я попытался найти спецификацию KTable API (https://kafka.apache.org/20/javadoc/org/apache/kafka/streams/kstream/KTable.html),, но не существует метода сортировки. Я также посмотрел эту статью (https://dzone.com/articles/how-to-order-streamed-dataframes), в которой предлагается реализовать сортировку через Processor API. Однако , Я проверяю, можно ли это сделать другим способом?

Answer 1

KafkaStream позволяет вам материализовать запрашиваемые государственные хранилища. Затем вы можете получить доступ только для чтения к магазину, вызвав метод kafkaStream#store().

Если вы определите постоянное хранилище, KafkaStreams будет использовать RocksDB для хранения ваших данных. Возвращенный экземпляр KeyValueIterator будет использовать итератор RocksDB, который позволит вам перебирать значения ключей в отсортированном виде Реализация итератора Rocks .

Пример:

    KafkaStreams streams = new KafkaStreams(topology, props);
    ReadOnlyKeyValueStore<Object, Object> store = streams.store("storeName", QueryableStoreTypes.keyValueStore());
    KeyValueIterator<Object, Object> iterator = store.all();