This page collects common usage pattern on how to use Kafka Streams Processor API or DSL.
Feel free to add your own code snippets and/or improve existing examples. |
Kafka Streams DSL aggregation natively support so-called "incremental" or "cumulative" aggregation functions (cf, https://en.wikipedia.org/wiki/Associative_property – also called "distributive functions" by Jim Gray in "Data Cube: A Relational Aggregation Operator Generalizing Group-By, Cross-Tab, and Sub-Totals") like count, sum, min, max. However, average function (a so-called "algebraic" function, cf. "Data Cube: A Relational Aggregation Operator Generalizing Group-By, Cross-Tab, and Sub-Totals") cannot be computed like this, but it can be composed of distributive functions, namely count and sum. Thus, it can be implemented in a two step approach:
class Tuple2<T1, T2> {
public T1 value1;
public T2 value2;
Tuple2(T1 v1, T2 v2) {
value1 = v1;
value2 = v2;
}
}
final KStreamBuilder builder = new KStreamBuilder();
// first step: compute count and sum in a single aggregation step and emit 2-tuples <count,sum> as aggregation result values
final KTable<String,Tuple2<Long,Long>> countAndSum = builder.stream("someInputTopic")
.groupByKey()
.aggregate(
new Initializer<Tuple2<Long, Long>>() {
@Override
public Tuple2<Long, Long> apply() {
return new Tuple2<>(0L, 0L);
}
},
new Aggregator<String, String, Tuple2<Long, Long>>() {
@Override
public Tuple2<Long, Long> apply(final String key, final Long value, final Tuple2<Long, Long> aggregate) {
++aggregate.value1;
aggregate.value2 += value;
return aggregate;
}
},
new Tuple2Serde()); // omitted for brevity
// second stop: compute average for each 2-tuple
final KTable<String,Double> average = countAndSum.mapValues(
new ValueMapper<Tuple2<Long, Long>, Double>() {
@Override
public Double apply(Tuple2<Long, Long> value) {
return value.value2 / (double) value.value1;
}
}); |
This pattern can also be applied to compute a windowed average or to compose other algebraic functions.
An application that uses aggregations can make better use of it's resources by removing records it no longer needs from its state stores. Kafka Streams makes this possible through the usage of tombstone records, which are records that contain a non-null key, and a null value. When Kafka Streams sees a tombstone record, it deletes the corresponding key from the state store, thus freeing up space.
The usage of tombstone records will become apparent in the example below, but it's important to note that any record with a null key will be internally dropped, and will not be seen by your aggregation. Therefore, it is necessary that your aggregation include the logic for recognizing when a record can be dropped from the state store, and by returning null when this condition is met.
Consider the following example. An airline wants to track the various stages of a customer's flight. For this example, a customer can be in one of 4 stages: booked, boarded, landed, and post-flight survey completed. Once the customer has completed the post-flight survey, the airline no longer needs to track the customer. Until then, the airline would like to know what stage the customer is in, and perform various aggregations on the customer's data. This can be accomplished using the following topology.
// customer flight statuses
final String BOOKED = "booked";
final String BOARDED = "boarded";
final String LANDED = "landed";
final String COMPLETED_FLIGHT_SURVEY = "survey";
// topics
final String SOURCE_TOPIC = "someInputTopic";
final String STORE_NAME = "someStateStore";
// topology
KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> stream = builder.stream(SOURCE_TOPIC);
KTable<String, String> customerFlightStatus = stream
.groupByKey()
.reduce(
new Reducer<String>() {
@Override
public String apply(String value1, String value2) {
if (value2.equals(COMPLETED_FLIGHT_SURVEY)) {
// we no longer need to keep track of this customer since
// they completed the flight survey. Create a tombstone
return null;
}
// keeping this simple for brevity
return value2;
}
}, STORE_NAME); |
Returning null in the Reducer, but only when the customer has completed their post-flight survey, allows us to perform aggregations until we no longer are interested in tracking this key. A tombstone is created by returning null, and the record is deleted from the state store immediately. We can verify this with the following tests:
// Simple test final String customerFlightNumber = "customer123_dl90210"; File stateDir = createStateDir(); // implementation omitted KStreamTestDriver driver = new KStreamTestDriver(builder, stateDir, Serdes.String(), Serdes.String()); driver.setTime(0L); // get a reference to the state store KeyValueStore<String, String> store = (KeyValueStore<String, String>) driver.context().getStateStore(STORE_NAME); // the customer has booked their flight driver.process(SOURCE_TOPIC, customerFlightNumber, BOOKED); store.flush(); assertEquals(BOOKED, store.get(customerFlightNumber)); // the customer has boarded their flight driver.process(SOURCE_TOPIC, customerFlightNumber, BOARDED); store.flush(); assertEquals(BOARDED, store.get(customerFlightNumber)); // the customer's flight has landed driver.process(SOURCE_TOPIC, customerFlightNumber, LANDED); store.flush(); assertEquals(LANDED, store.get(customerFlightNumber)); // the customer has filled out the post-flight survey, so we no longer need to track them // in the state store. make sure the key was deleted driver.process(SOURCE_TOPIC, customerFlightNumber, COMPLETED_FLIGHT_SURVEY); store.flush(); assertEquals(null, store.get(customerFlightNumber)); |
Finally, it's important to note how tombstones are forwarded downstream. Whether or a not a tombstone is visible to additional sub-topologies depends on which abstraction (e.g. KTable or KStream) a sub-topology uses for streaming it's input data. The following code snippets highlight these differences. Tombstones are visible in record streams, and it is common to filter them out before performing additional transformations (see below). However, in changelog streams, the tombstones are forwarded directly, and not visible when using operators like filter, mapValues, etc.
// tombstones are visible in a KStream and must be filtered out
KStream<String, String> subtopologyStream = customerFlightStatus
.toStream()
.filter((key, value) -> {
if (value == null) {
// tombstone! skip this
return false;
}
// other filtering conditions...
return true;
});
// tombstone forwarding is different in KTables. The filter below is not evaluated for a tombstone
KTable<String, String> subtopologyKtable = customerFlightStatus
.filter((key, value) -> {
// the tombstone never makes it here. no need to check for null
// other filtering conditions...
return true;
}); |