“WindowedBy Count KStream”抛出KStream异常

Question

我尝试将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-87ce 164d-c427-4d cf-aa76-aeeb 6 f 8 fc 943-stream thread-1”org . Apache . Kafka . streams . errors . streams Exception中出现异常:在进程中捕获到异常。taskId=0_0，processor = KSTREAM-SOURCE-000000000，topic=vehicle，partition=0，offset=160385位于org . Apache . Kafka . streams . processor . internals . stream task . process(stream task . Java:318)位于org . Apache . Kafka . streams . processor . internals . assignedstreamstasks . process(assignedstreamstasks . Java:94)位于org . Apache . Kafka . streams . processor . internals . task manager . process(task manager更改StreamConfig中的默认Serdes，或者通过方法参数提供正确的Serdes。

Answer 1

groupByKey（）使用默认序列化器：

组按键（）

将记录按其当前键分组到KGrouedStream中，同时保留原始值以及默认序列化器和反序列化器。

您必须使用组ByKey（序列化

以下内容应该可以解决问题:

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