KStream represents a continuous stream of Kafka records. It lets an application express transformations such as filtering, mapping, and aggregation through the Kafka Streams DSL instead of manually coordinating consumers, producers, offsets, and partitions. section 8.1.3
Architecturally, a KStream aligns with a Kafka topic used as an event source or sink. It represents the incoming and outgoing event flow, while KTable represents a stateful, table-like view backed by a changelog topic. section 8.3.2
KStream is useful when an application needs to:
- Filter events according to business conditions. section 8.1.3
- Map event values into a different representation. section 8.1.3
- Feed continuous events into stateful processing, including counts and time-windowed aggregations. section 8.1.3
Filtering and mapping orders
An application reads raw-orders into a KStream, keeps orders whose total is at least 50, maps each value to an order-and-customer JSON shape, and writes the result to processed-orders. section 8.1.3
Listing 8.1 OrderApp performing filtering and mapping of orders
public class OrderApp {
public static void main(String[] args) {
Properties props = new Properties(); // #1
...
StreamsBuilder builder = new StreamsBuilder(); // #2
KStream<String, String> ordersStream = builder.stream("raw-
➥orders"); // #3
KStream<String, String> processedOrders = ordersStream
.filter((key, value) -> { #4
Order order = Order.fromJson(value);
return order.getTotalAmount() >= 50;
})
.mapValues(value -> { // #5
Order order = Order.fromJson(value);
return String.format("{\"orderId\": \"%s\",
\"customerId\": \"%s\"}",
order.getOrderId(), order.getCustomerId());
});
processedOrders.to("processed-orders"); // #6
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
}
}
Inspect the compact Java topology showing stream creation, filtering, value mapping, output, and application startup. section 8.1.3
