KeyValueMapper maps one Kafka Streams record's key and value to a new KeyValue pair. Unlike ValueMapper, it can use the input key and produce a different key, making it useful when downstream records must be regrouped or repartitioned. section 6.4.1
Use the interface when the transformation needs to change record structure or routing:
- Implement apply(key, value) and return KeyValue.pair(newKey, newValue), rather than returning only a value. section 6.4.1
- Use KStream.map with KeyValueMapper when producing one transformed record for each input record. section 6.4.1
- Use flatMap with a KeyValueMapper when one input must emit zero or more records, such as one PurchasedItem record per purchase item. section 6.4.2
- Return new key and value objects instead of modifying the originals; Kafka Streams expects functions to be side-effect free. section 6.4.1
Rekeying a retail purchase
A mapper can inspect a customer ID, mask sensitive fields for IDs ending in "333", and return the store ID as the new key: `return KeyValue.pair(value.getStoreId(), builder.build());`. This supports later aggregation by store rather than by customer. section 6.4.1
Listing 6.9 KeyValueMapper
public class RetailKeyValueMapper implements
KeyValueMapper<String, RetailPurchase,
KeyValue<String, RetailPurchase>> {
@Override
public KeyValue<String, RetailPurchase> apply(String key,
RetailPurchase value) {
RetailPurchase.Builder builder = RetailPurchase.newBuilder();
if(key != null && key.endsWith("333")){ #1
builder.setCreditCardNumber("0000000000");
builder.setCustomerId("000000000");
}
return KeyValue.pair(value.getStoreId(), builder.build()); #2
}
}
Inspect the complete Java implementation showing the generic types, apply method, conditional masking, and changed key. section 6.4.1
Listing 6.10 KeyValueMapper returning a collection of PurchasedItem objects
KeyValueMapper<String, RetailPurchase,
Iterable<KeyValue<String, PurchasedItem>>> retailTransactionToPurchases =
(key, value) -> {
String zipcode = value.getZipCode(); #1
return value.getPurchasedItemsList().stream() #2
.map(purchasedItem ->
KeyValue.pair(zipcode, purchasedItem))
.collect(Collectors.toList());
}
See how a KeyValueMapper returns an Iterable of keyed PurchasedItem records for flatMap processing. section 6.4.2
