“WindowedBy Count KStream”抛出 StreamsException答案

【问题标题】："WindowedBy Count KStream" throws StreamsException“WindowedBy Count KStream”抛出 StreamsException
【发布时间】：2019-11-11 23:35:14
【问题描述】：

我试图将来自 KStream 的事件统计到时间段中：

    KStream<String, VehicleEventTO> stream = builder.stream("vehicle", Consumed.with(Serdes.String(), new JsonSerde<>(VehicleEventTO.class)));

    KStream<String, VehicleEventTO> streamWithKey = stream.selectKey((key, value) -> value.getId_vehicle().toString());

    KStream<String, Long> streamValueKey = streamWithKey.map((key, value) -> KeyValue.pair(key, value.getId_vehicle()));

    streamValueKey.groupByKey()
                  .windowedBy(TimeWindows.of(Duration.ofMinutes(10).toMillis()))
                  .count(Materialized.with(Serdes.String(), new JsonSerde<>(Long.class)));

我有这个例外：

线程异常 “test-app-87ce164d-c427-4dcf-aa76-aeeb6f8fc943-StreamThread-1” org.apache.kafka.streams.errors.StreamsException：异常捕获过程。 taskId=0_0，处理器=KSTREAM-SOURCE-0000000000，主题=车辆，分区=0，偏移量=160385 在 org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:318) 在 org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:94) 在 org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:409) 在 org.apache.kafka.streams.processor.internals.StreamThread.processAndMaybeCommit(StreamThread.java:964) 在 org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:832) 在 org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767) 在 org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736) 引起：org.apache.kafka.streams.errors.StreamsException：A 序列化程序（键： org.apache.kafka.common.serialization.ByteArraySerializer / 值： org.apache.kafka.common.serialization.ByteArraySerializer) 不是兼容实际的键或值类型（键类型：java.lang.String /值类型：java.lang.Long)。更改默认的 Serdes StreamConfig 或通过方法参数提供正确的 Serdes。

【问题讨论】：

标签： spring count apache-kafka apache-kafka-streams

【解决方案1】：

groupByKey() 使用默认的序列化器：

groupByKey()

按当前键将记录分组到一个 KGroupedStream 同时保留原始值和默认值序列化器和反序列化器。

您必须使用groupByKey(Serialized<K,V> serialized) 或groupByKey(Grouped<K,V> grouped)。

以下应该可以解决问题：

streamValueKey.groupByKey(Serialized.with(Serdes.String(), Serdes.Long()))
              .windowedBy(TimeWindows.of(Duration.ofMinutes(10).toMillis()))
              .count(Materialized.with(Serdes.String(), new JsonSerde<>(Long.class)));

【讨论】：