如何使用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()。