Stream Processing Topology
IntermediateA topology is a DAG of source, processor, and sink nodes; it is compiled from the high-level DSL (map, filter, join, aggregate) or the low-level Processor API.
Overview
A Kafka Streams topology is a Directed Acyclic Graph (DAG) of processing nodes. Source nodes read from Kafka topics, processor nodes transform/filter/aggregate records, and sink nodes write results to Kafka topics or state stores. You build topologies either through the high-level Streams DSL (fluent API using KStream, KTable, KGroupedStream) or the low-level Processor API (implement Processor<K,V> directly for custom logic). The DSL is idiomatic for most use cases — the Processor API is useful when you need fine-grained control over punctuation (time-based triggers), custom state stores, or side effects. Kafka Streams runs as a library inside your application — no separate cluster needed.
Building a topology with the Streams DSL
Use StreamsBuilder to define sources, transformations, and sinks. KStream represents an unbounded sequence of records; KTable represents a changelog stream with last-value-per-key semantics. Common operations: filter, map, flatMap, groupBy, aggregate, join, merge, peek, to (sink).
@Configuration
public class OrderStreamConfig {
@Bean
public KStream<String, OrderEvent> orderStream(StreamsBuilder builder) {
// Source: read from "order-events" topic
KStream<String, OrderEvent> stream = builder.stream(
"order-events",
Consumed.with(Serdes.String(), orderEventSerde())
);
// Filter: only PLACED orders
KStream<String, OrderEvent> placed = stream
.filter((key, event) -> event.getType() == OrderEventType.PLACED);
// Transform: enrich with metadata
KStream<String, EnrichedOrder> enriched = placed
.mapValues(event -> new EnrichedOrder(
event.getOrderId(),
event.getCustomerId(),
Instant.now()
));
// Branch: split by order value
Map<String, KStream<String, EnrichedOrder>> branches = enriched
.split(Named.as("branch-"))
.branch((k, v) -> v.getTotal().compareTo(new BigDecimal("1000")) > 0,
Branched.as("high-value"))
.defaultBranch(Branched.as("standard"));
// Sink: write to separate topics
branches.get("branch-high-value").to("high-value-orders");
branches.get("branch-standard").to("standard-orders");
return stream;
}
}Aggregations and KTable
Grouping and aggregating produce KTable results — a compacted changelog of the latest aggregate value per key. count() is the simplest aggregation; aggregate() builds arbitrary accumulators. Results are materialised to state stores (RocksDB) and can be queried via Interactive Queries.
// Order count per customer — produces a KTable
StreamsBuilder builder = new StreamsBuilder();
KTable<String, Long> orderCountPerCustomer = builder
.stream("order-events", Consumed.with(Serdes.String(), orderEventSerde()))
.filter((k, v) -> v.getType() == OrderEventType.PLACED)
.groupBy(
(key, event) -> event.getCustomerId(), // rekey by customerId
Grouped.with(Serdes.String(), orderEventSerde())
)
.count(Materialized.as("order-count-store")); // named store → queryable
// Revenue per customer — custom aggregator
KTable<String, BigDecimal> revenuePerCustomer = builder
.stream("order-events", Consumed.with(Serdes.String(), orderEventSerde()))
.filter((k, v) -> v.getType() == OrderEventType.PLACED)
.groupBy((k, v) -> v.getCustomerId(),
Grouped.with(Serdes.String(), orderEventSerde()))
.aggregate(
() -> BigDecimal.ZERO, // initializer
(customerId, event, total) ->
total.add(event.getTotal()), // aggregator
Materialized.<String, BigDecimal, KeyValueStore<Bytes, byte[]>>as(
"revenue-store")
.withValueSerde(bigDecimalSerde())
);
// Sink KTable as changelog topic
orderCountPerCustomer.toStream().to("customer-order-counts");Stream-Table join and topology inspection
Joining a KStream with a KTable enriches each stream record with the current table value for the same key — a common pattern for enriching events with reference data. Use Topology.describe() during development to visualise the processing graph and verify the DAG structure.
// KStream-KTable join: enrich order events with customer data
KTable<String, Customer> customers = builder.table(
"customers",
Consumed.with(Serdes.String(), customerSerde())
);
KStream<String, EnrichedOrderEvent> enriched = orderStream
.join(
customers,
(order, customer) -> new EnrichedOrderEvent(
order.getOrderId(), customer.getName(), customer.getEmail(), order.getTotal()
)
// join key = order.getCustomerId() (must match table key)
);
enriched.to("enriched-order-events");
// Inspect topology DAG (log during startup/testing)
Topology topology = builder.build();
System.out.println(topology.describe());
// Outputs: Sub-topology: 0
// Source: KSTREAM-SOURCE-0000000000 (topics: [order-events])
// Processor: KSTREAM-JOIN-0000000002 (stores: [customers])
// Source: KTABLE-SOURCE-0000000001 (topics: [customers])
// Sink: KSTREAM-SINK-0000000003 (topic: enriched-order-events)Key Points to Remember
- 1A topology is a DAG: source nodes (from topics) → processor nodes (transform/filter/aggregate) → sink nodes (to topics)
- 2KStream = unbounded sequence of records; KTable = last value per key (changelog semantics)
- 3groupBy() + count()/aggregate() produces a KTable materialised in a named RocksDB state store
- 4KStream-KTable join enriches each stream record with the latest value for the matching key in the table
- 5Kafka Streams runs as a library in your application — no separate cluster or worker nodes needed
- 6topology.describe() prints the full DAG structure — useful for debugging and understanding the processing graph
Interview Questions
Sign in to ask AriaWhat is the difference between a KStream and a KTable in Kafka Streams?
How does groupBy().count() work internally and where is the result stored?
When would you use a KStream-KTable join vs a KStream-KStream join?
What is a sub-topology in Kafka Streams and why might a topology be split into multiple sub-topologies?
How does Kafka Streams achieve fault tolerance without a separate cluster?
Ask Aria about Stream Processing Topology
Your personal AI tutor — ask anything about this concept
Revision Status
Personal Notes
Sign in to save personal notes for this topic.
Discussion
Sign in to join the discussion.