Kafka Streams: Stream Processing

Kafka Streams is a client library for building applications and microservices that process and analyze data stored in Kafka. It reads from input topics, applies transformations, and writes results back to Kafka: all without a separate processing cluster. A Streams application is just a regular Java app.

Kafka

What is Kafka Streams
KStream vs KTable
Building a Streams App
Stateless Operations (filter, map, flatMap)
Stateful Operations (groupBy, count, aggregate)
Windowing

What is Kafka Streams

Kafka Streams is part of the Apache Kafka project. Unlike Storm, Spark Streaming, or Flink, it does not require a dedicated cluster: it runs as threads inside your application process. Kafka handles durability, replication, and partitioning.

Reads from one or more Kafka topics and writes results to Kafka topics.
Supports exactly-once processing semantics.
Scales by adding more application instances: Kafka Streams balances partitions across instances automatically.
Provides a high-level DSL (Streams DSL) and a low-level Processor API.

Add the dependency: kafka-streams (same version as your Kafka broker). No separate broker is needed: just a running Kafka cluster.

KStream vs KTable

The two core abstractions are KStream and KTable.

Abstraction	Represents	Analogy
`KStream`	An unbounded stream of records: every record is an independent event	Like a database INSERT log
`KTable`	A changelog stream: each record is an upsert (update or insert) keyed by the record key	Like a database table with updates
`GlobalKTable`	A KTable replicated fully across all instances: useful for enrichment lookups	Like a broadcast dimension table

A KTable keeps only the latest value per key. Earlier values for the same key are replaced when a new record arrives. A KStream treats every record as a new, independent event regardless of key.

Building a Streams App

A minimal Kafka Streams application consists of a StreamsBuilder, a topology, and a KafkaStreams instance.

Java

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.common.serialization.Serdes;
import java.util.Properties;

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "word-count-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

StreamsBuilder builder = new StreamsBuilder();

// Read from input topic
KStream<String, String> textLines = builder.stream("text-input");

// Count words and write to output topic
textLines
    .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
    .groupBy((key, word) -> word)
    .count()
    .toStream()
    .to("word-count-output", Produced.with(Serdes.String(), Serdes.Long()));

Topology topology = builder.build();
KafkaStreams streams = new KafkaStreams(topology, props);
streams.start();

// Graceful shutdown
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

The application.id is also used as the consumer group ID. All instances with the same application.id collaborate on processing.

Stateless Operations (filter, map, flatMap)

Stateless operations process each record independently: no state store is needed.

Java

KStream<String, String> orders = builder.stream("orders");

// filter — keep only records matching a predicate
KStream<String, String> highValueOrders = orders
    .filter((key, value) -> value.contains("\"amount\":") && parseAmount(value) > 100);

// filterNot — drop matching records
KStream<String, String> validOrders = orders
    .filterNot((key, value) -> value.contains("\"status\":\"cancelled\""));

// map — transform key and/or value
KStream<String, String> upperCased = orders
    .map((key, value) -> KeyValue.pair(key.toUpperCase(), value.toUpperCase()));

// mapValues — transform value only (key unchanged)
KStream<String, Order> parsed = orders
    .mapValues(value -> objectMapper.readValue(value, Order.class));

// flatMap — one record → zero or more records
KStream<String, String> items = orders
    .flatMap((key, value) -> extractLineItems(key, value));

// flatMapValues — one value → multiple values, key unchanged
KStream<String, String> words = orders
    .flatMapValues(value -> Arrays.asList(value.split(" ")));

// selectKey — assign a new key
KStream<String, String> byCustomer = orders
    .selectKey((key, value) -> extractCustomerId(value));

Stateful Operations (groupBy, count, aggregate)

Stateful operations maintain state in a state store (backed by a local RocksDB instance and a Kafka changelog topic). They always start with a groupByKey() or groupBy() call.

Java

KStream<String, String> events = builder.stream("page-views");

// count per key
KTable<String, Long> viewCounts = events
    .groupByKey()
    .count(Materialized.as("view-count-store"));

// count after re-keying
KTable<String, Long> countByPage = events
    .groupBy((key, value) -> KeyValue.pair(extractPage(value), value))
    .count();

// aggregate — custom accumulator
KTable<String, Double> totalRevenue = builder
    .<String, String>stream("sales")
    .groupByKey()
    .aggregate(
        () -> 0.0,                          // initializer
        (key, value, aggValue) ->            // aggregator
            aggValue + parseAmount(value),
        Materialized.<String, Double, KeyValueStore<Bytes, byte[]>>as("revenue-store")
            .withValueSerde(Serdes.Double())
    );

// reduce — special case of aggregate for same type
KTable<String, String> latest = events
    .groupByKey()
    .reduce((aggValue, newValue) -> newValue); // keep latest

State stores are queryable via the Interactive Queries API, allowing you to expose the current state of a KTable through a REST endpoint.

Windowing

Windowing groups records into time-bounded buckets for stateful operations like count or aggregate. Kafka Streams supports several window types.

Window Type	Description	Example
Tumbling	Fixed-size, non-overlapping windows	Count events per 1-minute window
Hopping	Fixed-size, overlapping windows that advance by a smaller step	5-minute window advancing every 1 minute
Sliding	Windows that slide continuously on every new event	Events within 5 minutes of each other
Session	Activity-based windows: merged when events are within a gap interval	User sessions with 30 min inactivity gap

Java

import org.apache.kafka.streams.kstream.*;
import java.time.Duration;

KStream<String, String> clicks = builder.stream("user-clicks");

// Tumbling window — count clicks per user per 5 minutes
KTable<Windowed<String>, Long> tumblingCounts = clicks
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count();

// Hopping window — 10 min window, advances every 2 min
KTable<Windowed<String>, Long> hoppingCounts = clicks
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(10), Duration.ofSeconds(30))
        .advanceBy(Duration.ofMinutes(2)))
    .count();

// Session window — sessions with 5 min inactivity gap
KTable<Windowed<String>, Long> sessionCounts = clicks
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(5)))
    .count();

// Write windowed results to output topic
tumblingCounts
    .toStream()
    .map((windowedKey, count) -> KeyValue.pair(windowedKey.key(), count))
    .to("click-counts-output", Produced.with(Serdes.String(), Serdes.Long()));

Late-arriving records outside the grace period are dropped. Use ofSizeAndGrace() to allow a grace period for out-of-order events rather than ofSizeWithNoGrace().

Contents

What is Kafka Streams

KStream vs KTable

Building a Streams App

Stateless Operations (filter, map, flatMap)

Stateful Operations (groupBy, count, aggregate)

Windowing

Related articles