如何使用KeyValueMapper将输入KStream <String，String>映射到<String，CarClass>？

Question

接收我只想在CarClass上映射的Json数据，并想创建新的流，但映射方法不允许我映射自定义数据类型 类型KStream中的方法映射（KeyValueMapper>）不适用于参数（新的KeyValueMapper>（）{}）？

Answer 1

从http://docs.confluent.io/current/streams/developer-guide.html#stateless-transformations：

示例改变从byte[]到Integer值类型。对于String到CarClass是相同的。

KStream<byte[], String> stream = ...; 

// Java 8+ example, using lambda expressions 
// Note how we change the key and the key type (similar to 
`selectKey`) 
// as well as the value and the value type. 
KStream<String, Integer> transformed = stream.map(
    (key, value) -> KeyValue.pair(value.toLowerCase(), value.length())); 

// Java 7 example 
KStream<String, Integer> transformed = stream.map(
    new KeyValueMapper<byte[], String, KeyValue<String, Integer>>() { 
    @Override 
    public KeyValue<String, Integer> apply(byte[] key, String value) { 
     return new KeyValue<>(value.toLowerCase(), value.length()); 
    } 
    }); 

不过，如果你希望只修改这个值，我会建议使用mapValues()而不是map()。