Создание SerDes для оконных данных в потоках Kafka

Question

У меня возникли проблемы с созданием SerDes для использования с данными, которые я собираю, и мне нужно отправить их в другую тему через .to (), однако мне нужно создать SerDes для оконных данных, и я не уверен, как это сделать.

Answer 1

Мы можем создать Serilizer и DeSerilizer для оконных данных следующими способами.

StringSerializer stringSerializer = new StringSerializer();
StringDeserializer stringDeserializer = new StringDeserializer();
Serde<String> stringSerde = Serdes.serdeFrom(stringSerializer,stringDeserializer);

WindowedSerializer<String> windowedSerializer = new WindowedSerializer<>(stringSerializer);
WindowedDeserializer<String> windowedDeserializer = new WindowedDeserializer<>(stringDeserializer);
Serde<Windowed<String>> windowedSerde = Serdes.serdeFrom(windowedSerializer,windowedDeserializer);

Использование Serilizer / DeSerilizer в оконных данных приведено ниже.

KStream<String,StockTransaction> transactionKStream =  kStreamBuilder.stream(stringSerde,transactionSerde,"stocks");

transactionKStream.map((k,v)-> new KeyValue<>(v.getSymbol(),v))
                              .through(stringSerde, transactionSerde,"stocks-out")
                              .groupBy((k,v) -> k, stringSerde, transactionSerde)
                              .aggregate(StockTransactionCollector::new,
                                   (k, v, stockTransactionCollector) -> stockTransactionCollector.add(v),
                                   TimeWindows.of(10000),
                                   collectorSerde, "stock-summaries")
                    .to(windowedSerde,collectorSerde,"transaction-summary");

Iпредложил бы вам пройти через следующий контент, чтобы получить больше информации.

https://www.programcreek.com/java-api-examples/index.php?api=org.apache.kafka.streams.kstream.internals.WindowedSerializer