在Kafka Streams中指定映射操作上的serdes

Question

我很抱歉，如果这是一个重复的帖子，但我一直在寻找答案，到目前为止什么也没有得到。

我需要的是在更改Kafka Streams中的键类型的映射操作期间指定serdes。原始KStream有一个字符串类型的键和avro（通用记录）值，但我需要将其重新映射到avro键和值。大致如下：

KStream<String, GenericRecord> inputStream = builder.stream("someTopic");
KStream<GenericRecord, GenericRecord> rekeyedStream = inputStream.map((key, value) -> {
   GenericRecord newKey = new GenericData.Record(someSchema);
   ...
   return new KeyValue(newKey, newValue);
});

我相信我需要指定serde，因为类型正在改变，但是我在地图操作符上找不到方法。当从主题读取、分组或写回主题时，我们通常可以做如下操作来覆盖默认serde：

KStream<GenericRecord, GenericRecord> stream = builder.stream("someTopic", 
    Consumed.with(keySerde, valueSerde));

KGroupedStream<GenericRecord, GenericRecord> groupedStream = 
    inputStream.groupBy((key, value)->somethingThatChangesTheKey(), 
    Grouped.with(newKeySerde, newValueSerde));

inputStream.to("someTopic", Produced.with(keySerde,valueSerde));

然而，当类型发生变化时，我完全不知道如何在地图中指定serdes，在这种特殊情况下，我不能使用我的应用程序的默认serdes。

我最接近找到解决方案的地方就是这里的这篇帖子，但我担心被接受的回复告诉OP，他需要指定SERDE，但不是如何在地图上完成（至少据我所知，我可能弄错了）。

如有任何见解，将不胜感激。

Answer 1

无法在map（）上指定新的serde，因为map（）不需要serde。map（）操作符本身获取一个输入对象并生成一个输出对象，但它从不序列化或反序列化任何消息。

只有读取或写入Kafka主题的运算符才允许设置serde，因为只有这些运算符才会使用serde。

我不清楚在map（）运算符上设置一个serde要完成什么。你能详细说明一下吗？