我有数据流作为
StreamsBuilder streamsBuilder = new StreamsBuilder();
streamsBuilder.stream("events")
.groupByKey()
.windowedBy(TimeWindows.of(Duration.ofMillis(10000)))
.count()
.toStream()
.to("output");
但我收到错误
ClassCastException while producing data to topic output. A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: org.apache.kafka.common.serialization.LongSerializer) is not compatible to the actual key or value type (key type: org.apache.kafka.streams.kstream.Windowed / value type: java.lang.Long). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters (for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with `Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).
windowedBy
加上count()
的结果是类型为<Windowed<String>, Long>
的键值对,因此您需要通过to()
参数在Produced
中设置不同的serde。默认情况下,将使用配置中的 Serdes,看起来您设置为 StringSerde/StringSerde
并且这些显然与输出主题键/值类型不匹配。
Kafka Streams 附带用于窗口类型的内置 Serdes,您可以通过
Serdes
工厂类获得。
可以尝试下面的方法
.count()
.toStream()
.map((key, value) -> new KeyValue<>(key.key() + "@" + key.window().start() + "->" + key.window().end(), value))
.to("output");