我正在尝试使用以下代码根据键生成计数，此代码基于字数统计示例。奇怪的是，如果 mapValues 函数返回一个字符串，那么 groupBy 就像注释行中提到的那样工作，但是当我发送一个字符串的密钥对作为键和 GenericRecord 作为值时。
final Serde<String> stringSerde = Serdes.String();
final Serde<Long> longSerde = Serdes.Long();
final Map<String, String> serdeConfig = Collections.singletonMap("schema.registry.url","http://localhost:8081");
stringSerde.configure(serdeConfig, true); // `true` for record keys
final Serde<GenericRecord> valueGenericAvroSerde = new GenericAvroSerde();
valueGenericAvroSerde.configure(serdeConfig, false); // `false` for record values
StreamsBuilder builder = new StreamsBuilder();
KStream<String, GenericRecord> textLines =
builder.stream("ora-query-in",Consumed.with(stringSerde, valueGenericAvroSerde));
final KTable<String, Long> wordCounts = textLines
.mapValues(new ValueMapperWithKey<String, GenericRecord, KeyValue<String, GenericRecord>>() {
@Override
public KeyValue<String, GenericRecord> apply(String arg0, GenericRecord arg1) {
return new KeyValue<String, GenericRecord>(arg1.get("KEY_FIELD").toString(),arg1);
}
})
// .groupBy((key, value) -> value) //THIS WORKS if value is STRING
// .groupBy((key, value) -> key) //DOES NOT WORK EITHER
.groupByKey() //THIS does nothing
.count();
wordCounts.toStream().to("test.topic.out",Produced.with(stringSerde, longSerde));
我在配置中遗漏了什么吗
streamsConfiguration.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());