Home/Learn/Apache Kafka/KStream & KTable

KStream & KTable

Intermediate

Kafka Streams

KStream represents an unbounded event log; KTable represents a changelog-backed materialised view (latest value per key); GlobalKTable replicates fully to every instance.

Overview

KStream and KTable are the two fundamental abstractions in Kafka Streams. A KStream is an unbounded, immutable event log — every record is an independent event (like a click or a payment). A KTable is a changelog stream interpreted as a materialised view: each record upserts the current value for its key, and the latest value per key is what matters (like a user profile). The distinction determines join semantics: KStream-KTable joins enrich each event with the latest snapshot of the table; KStream-KStream joins correlate two event streams in a time window. GlobalKTable is a special variant that replicates the entire table to every Streams instance, enabling efficient joins without co-partitioning.

KStream — event-by-event processing

KStream processes each record independently. Use map, filter, flatMap, and peek for stateless transformations; branch for routing.

Java — KStream filter, map, branch

StreamsBuilder builder = new StreamsBuilder();

// Source: read raw order events
KStream<String, OrderEvent> orders =
    builder.stream("order-events",
        Consumed.with(Serdes.String(), orderEventSerde));

// Stateless transforms
KStream<String, OrderEvent> validOrders = orders
    .filter((key, order) -> order.getAmount().compareTo(BigDecimal.ZERO) > 0)
    .mapValues(order -> {
        order.setStatus("VALIDATED");
        return order;
    });

// Branch: route to different topics by value
Map<String, KStream<String, OrderEvent>> branches = validOrders.split()
    .branch((key, order) -> "EXPRESS".equals(order.getShipping()),
            Branched.withConsumer(s -> s.to("express-orders")))
    .defaultBranch(Branched.withConsumer(s -> s.to("standard-orders")));

// Peek for side effects (logging, metrics) without mutating
validOrders.peek((key, order) ->
    log.info("Processing order {} amount {}", key, order.getAmount()));

KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.start();

KTable — materialised view and changelog

KTable treats the latest record per key as current state. Internally backed by a RocksDB state store with a changelog topic for fault tolerance.

Java — KTable with KStream-KTable join

StreamsBuilder builder = new StreamsBuilder();

// KTable: each record upserts current user profile
KTable<String, UserProfile> users =
    builder.table("user-profiles",
        Consumed.with(Serdes.String(), userProfileSerde),
        Materialized.<String, UserProfile, KeyValueStore<Bytes, byte[]>>as("users-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(userProfileSerde));

// KStream-KTable join: enrich each order event with current user profile
KStream<String, OrderEvent> orders = builder.stream("order-events");

KStream<String, EnrichedOrder> enriched = orders.join(
    users,
    (order, user) -> new EnrichedOrder(order, user.getEmail(), user.getTier()),
    Joined.with(Serdes.String(), orderEventSerde, userProfileSerde)
);

enriched.to("enriched-orders");

// Interactive query: read the KTable state store directly
ReadOnlyKeyValueStore<String, UserProfile> store =
    streams.store(StoreQueryParameters.fromNameAndType(
        "users-store", QueryableStoreTypes.keyValueStore()));
UserProfile profile = store.get("user-123");

GlobalKTable — broadcast reference data

GlobalKTable replicates the entire table to every application instance regardless of partition assignment, removing the co-partitioning requirement for joins.

Java — GlobalKTable join without co-partitioning

StreamsBuilder builder = new StreamsBuilder();

// GlobalKTable is fully replicated to every instance
// Use for small reference data (products, categories, configs)
GlobalKTable<String, Product> products =
    builder.globalTable("product-catalog",
        Consumed.with(Serdes.String(), productSerde));

KStream<String, OrderItem> orderItems = builder.stream("order-items");

// GlobalKTable join: key extractor maps stream record to lookup key
KStream<String, EnrichedOrderItem> enriched = orderItems.join(
    products,
    (orderItemKey, orderItem) -> orderItem.getProductId(),  // FK lookup key
    (orderItem, product) -> new EnrichedOrderItem(orderItem, product.getName(),
                                                   product.getCategory())
);

// KTable vs GlobalKTable:
// KTable: partitioned, co-partitioning required, lower memory per instance
// GlobalKTable: fully replicated, no co-partitioning needed, higher memory
// Use GlobalKTable when: table is small (<100 MB) and joins are cross-partition

Key Points to Remember

1KStream = append-only event log; every record is independent. KTable = changelog; only the latest value per key matters.
2KStream-KTable joins look up the current table snapshot for each arriving stream event — useful for enrichment.
3KStream-KStream joins require a time window because both sides are unbounded streams.
4KTable state is backed by RocksDB locally and a changelog Kafka topic for fault tolerance and replication.
5GlobalKTable replicates fully to every instance — safe for small lookup tables; avoid for large datasets.
6Co-partitioning is required for KStream-KTable and KStream-KStream joins (same partition count, same partitioner).

Interview Questions

What is the difference between a KStream and a KTable in Kafka Streams?

EasyConfluent

When would you use a GlobalKTable instead of a KTable?

MediumLinkedIn

What is co-partitioning and why is it required for KStream-KTable joins?

HardUber

How does Kafka Streams recover KTable state after a node restart?

HardNetflix

What happens when a null value is written to a KTable topic?

MediumAmazon

Ask Aria about KStream & KTable

Your personal AI tutor — ask anything about this concept

Revision Status

Personal Notes

Discussion

Loading discussion…

Kafka Streams Basics

Stream Processing Topology