代码之家 › 专栏 › 技术社区 › sobychacko

KTable状态存储持久性

apache-kafka-streams

sobychacko · 技术社区 · 6 年前

如果在具体化ktable时使用持久存储,那么在应用程序重新启动时,状态存储是否持久?例如,如果我使用以下内容:

StreamsBuilder builder = new StreamsBuilder();
KeyValueBytesStoreSupplier storeSupplier =      Stores.persistentKeyValueStore("queryable-store-name");
 KTable<Long,String> table = builder.table(
   "foo",
   Materialized.as(storeSupplier)
               .withKeySerde(Serdes.Long())
               .withValueSerde(Serdes.String())

状态存储“可查询存储名称”是否可以通过重新启动时以前运行的状态访问?比方说,我将50条记录发送到主题foo,它在状态存储中实现。然后重新启动应用程序,状态存储区中还有50条记录吗?如果没有,有没有办法做到这一点?

谢谢!

1 回复 | 直到 6 年前

Matus Cimerman 6 年前

application-id

final KStream<CustomerKey, ViewPage> viewPagesStream=builder.stream(INPUT_TOPIC);

final KTable<Windowed<ViewPageCountKey>,Long>uniqueViewPageCount=viewPagesStream
        .map((key,value)->{
            ViewPageCountKey newKey=new ViewPageCountKey(key.getProjectId(),value.getUrl());
            return new KeyValue<>(newKey,value);
        })
        .filter((key,value)->key!=null)
        .groupByKey()
        .count(TimeWindows.of(WINDOW_SIZE).advanceBy(WINDOW_ADVANCE),STORE_NAME);

推荐文章

Stella · Kafka流处理期间处理数据时出现异常

7 年前

Stella · kafka流中的处理器节点

7 年前

mukesh210 · 卡夫卡流:Lib Rocks数据库上的UnsatifiedLinkError

7 年前

el323 · Kafka流:通过Json日志中的键进行分组

7 年前

Suchita · 如何使用Avro反序列化器和Avro序列化器为KafkaStreams编写Junit测试用例

7 年前

Cemo · 如何获得有关GlobalKTable状态存储更新的通知?

7 年前

px5x2 · Kafka将GlobalKTable同步流传输到应用程序

7 年前

Dth · 加工机为Kafka流中的单个任务多次调用init()

7 年前

nattyddubbs · Kafka Streams多实例设计含义

7 年前

dvlcis · 如何使用卡夫卡流窗口为烛台图生成创建一条记录

7 年前