为 Kafka 流上的窗口数据创建 SerDes答案

【问题标题】：Creating a SerDes for Windowed Data on Kafka Streams为 Kafka 流上的窗口数据创建 SerDes
【发布时间】：2019-10-01 19:09:27
【问题描述】：

我在创建用于我正在聚合的数据的 SerDes 时遇到了一些麻烦，并且需要通过“.to()”发送到另一个主题，但是，我需要为窗口数据创建一个 SerDes，我不知道该怎么做。

【问题讨论】：

标签： java apache-kafka aggregation apache-kafka-streams serialization

【解决方案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");

我建议您阅读以下内容以获取更多信息。

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

【讨论】：

根据kafka.apache.org/20/javadoc/org/apache/kafka/streams/kstream/…，我假设没有名为“Windowed(De)Serializer”的类，因为我想使用翻滚窗口，所以我必须使用 TimeWindowedSerializer？
WindowedDeserializer 是自定义解串器，您可以在github.com/axbaretto/kafka/blob/master/streams/src/main/java/… 找到它。是的，你必须使用 TimeWindowedSerializer
@RohitYadav 我建议您更新答案以使用`TimeWindowedSerde`...
这样的类型似乎是一致的： WindowedSerializer windowedSerializer = new TimeWindowedSerializer(new StringSerializer()); Deserializer> windowedDeserializer = new TimeWindowedDeserializer(new StringDeserializer()); Serde> windowedSerde = Serdes.serdeFrom(windowedSerializer, windowedDeserializer);