2.1 Core Abstractions: KStream, KTable & GlobalKTable

Module Overview

Goal: One authoritative place for stream/table semantics.

What You'll Learn:

• KStream semantics (event stream)
• KTable semantics (changelog / latest value)
• Update vs insert behavior
• GlobalKTable vs KTable differences
• When to use each abstraction
• The duality of streams and tables

What's NOT Covered Here:

• Joins implementation (covered in Module 8.1)
• State stores (covered in Module 6.1)
• Windowing (covered in Module 5.2)
• Specific operations (covered in Modules 4 and 5)

Key Insight: Understanding these three abstractions is fundamental — everything else in Kafka Streams builds on them.

1. The Fundamental Concept

Streams and Tables Are Dual

STREAM (KStream):
└─ Sequence of immutable events
└─ "What happened?"
└─ Every record is important

TABLE (KTable):
└─ Current state, mutable
└─ "What is the current value?"
└─ Only latest value per key matters

Key Insight: Streams and tables are two ways of looking at the same data:

A stream can be materialized into a table
A table can be converted into a stream of changes

Real-World Analogy: Pizza Delivery Service

Imagine you're running a pizza delivery service:

KStream = Order Slips Flying Into Your Kitchen

Every order slip that arrives:
├─ Order #101: Large Pepperoni
├─ Order #102: Medium Margherita  
├─ Order #103: Cancel order #101
└─ Order #104: Large Veggie

Every event matters individually!
Each slip is a separate, independent event.

KTable = Whiteboard Showing Current Status

Your whiteboard shows latest state only:

┌────────────────────────────────┐
│ Order Status Board             │
├────────────────────────────────┤
│ Order #101: Cancelled          │
│ Order #102: In Preparation     │
│ Order #104: Pending            │
└────────────────────────────────┘

You only care about the current status of each order.
Old states are overwritten.

The Core Difference:

• KStream: Every slip (event) is kept — you see all activity
• KTable: Only current status per order — latest value only

2. KStream: Event Stream

Definition

A KStream represents an unbounded sequence of immutable events.

Think of KStream as a log of everything that happened:

Time →
┌────────────────────────────────────────────→
│
│ Event1  Event2  Event3  Event4  Event5 ... ∞
│
└────────────────────────────────────────────→

Each event is independent
Order matters
All events preserved

Characteristics

KSTREAM PROPERTIES:

1. Immutable Events
   └─ Each record is an independent fact
   └─ Records never overwrite each other

2. Insert Semantics
   └─ New records are always appended
   └─ No concept of "update"

3. All History Preserved
   └─ Every record is kept (subject to retention)
   └─ You see all events, not just latest

4. Keyed or Unkeyed
   └─ Records may or may not have keys
   └─ Keys used for partitioning

Example: User Click Stream

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

Topic "user-clicks":
┌────────────────────────────────────────────────┐
│ Key: user1 | Value: {page: "home", time: 10:00}
│ Key: user2 | Value: {page: "products", time: 10:01}
│ Key: user1 | Value: {page: "cart", time: 10:02}
│ Key: user1 | Value: {page: "checkout", time: 10:03}
│ Key: user2 | Value: {page: "home", time: 10:04}
└────────────────────────────────────────────────┘

All 5 events are independent
user1 appears 3 times (3 separate events)
Order preserved within partition

Semantics: "User1 clicked home, then cart, then checkout"

Use Cases for KStream

✅ Event logging
   Example: User activity, clickstreams

✅ Transactions
   Example: Payment events, order placed

✅ Time-series data
   Example: Sensor readings, metrics

✅ Audit trails
   Example: All changes, never overwrite

✅ When every occurrence matters
   Example: Count ALL clicks, not just latest

3. KTable: Changelog / Latest Value

Definition

A KTable represents a changelog where only the latest value per key matters.

Think of KTable as a database table:

Key: user1
├─ Version 1: {name: "Alice", age: 25}
├─ Version 2: {name: "Alice", age: 26}  ← overwrites
└─ Current:   {name: "Alice", age: 26}  ← only this matters

Characteristics

KTABLE PROPERTIES:

1. Mutable State
   └─ Each record updates previous value for that key
   └─ Latest value wins

2. Update Semantics
   └─ New records overwrite old ones (same key)
   └─ Like SQL UPDATE statement

3. Latest Value Per Key
   └─ Historical values discarded
   └─ You see current snapshot

4. Always Keyed
   └─ Keys are mandatory
   └─ Key determines which value to update

Example: User Profile

KTable<String, UserProfile> users = builder.table("user-profiles");

Topic "user-profiles" (compacted):
┌────────────────────────────────────────────────┐
│ Key: user1 | Value: {name: "Alice", age: 25}
│ Key: user2 | Value: {name: "Bob", age: 30}
│ Key: user1 | Value: {name: "Alice", age: 26}  ← Update!
│ Key: user1 | Value: null                       ← Delete!
│ Key: user2 | Value: {name: "Bob", age: 31}    ← Update!
└────────────────────────────────────────────────┘

Current KTable State:
├─ user1: null (deleted)
└─ user2: {name: "Bob", age: 31}

Semantics: "What is the current profile for each user?"

Update vs Insert Behavior

KSTREAM (Insert):
Key: user1 → Event1
Key: user1 → Event2  ← Both kept
Key: user1 → Event3  ← All three exist

Result: 3 records for user1


KTABLE (Update):
Key: user1 → Value1
Key: user1 → Value2  ← Overwrites Value1
Key: user1 → Value3  ← Overwrites Value2

Result: 1 record for user1 (Value3)

Tombstone Records (Deletes)

KTable supports deletion via null values:

Key: user1 → {name: "Alice"}  ← Record exists
Key: user1 → null             ← Record deleted (tombstone)

After tombstone:
└─ user1 no longer in table
└─ Subsequent reads return null

Important: Tombstones are how you delete from a KTable.

Use Cases for KTable

✅ Current state / snapshots
   Example: User profiles, product catalog

✅ Reference data
   Example: Currency exchange rates

✅ Latest status
   Example: Order status, inventory levels

✅ Aggregations
   Example: Running totals (covered in Module 5.1)

✅ When only current value matters
   Example: "What is user1's current age?" not "How many times did age change?"

4. Stream-Table Duality

Converting Between Them

STREAM → TABLE:
└─ Aggregate/reduce stream into latest values
└─ KStream.toTable() or KStream.groupByKey().reduce()

TABLE → STREAM:
└─ Convert changelog into stream of updates
└─ KTable.toStream()

Example: Stream to Table

INPUT STREAM (clicks):
[user1, click1]
[user2, click2]
[user1, click3]
[user1, click4]

→ COUNT BY USER (aggregate)

OUTPUT TABLE (click_counts):
user1 → 3
user2 → 1

Table represents current count for each user

Example: Table to Stream

INPUT TABLE (user_profiles):
user1 → {name: "Alice", age: 25}
user2 → {name: "Bob", age: 30}
user1 → {name: "Alice", age: 26}  ← Update

→ CONVERT TO STREAM (toStream)

OUTPUT STREAM (profile_changes):
[user1, {name: "Alice", age: 25}]
[user2, {name: "Bob", age: 30}]
[user1, {name: "Alice", age: 26}]  ← All changes visible

Stream shows all update events

5. GlobalKTable: Replicated Table

What is GlobalKTable?

A GlobalKTable is like a KTable, but fully replicated to every Streams instance.

Analogy: If a KTable is like a local phone book split across cities, then a GlobalKTable is like having the full phone book in every city — everywhere, all the time.

KTABLE (partitioned):
Instance 1: Partition 0, 1
Instance 2: Partition 2, 3
└─ Each instance sees partial data

GLOBALKABLE (replicated):
Instance 1: ALL partitions (0, 1, 2, 3)
Instance 2: ALL partitions (0, 1, 2, 3)
└─ Each instance sees complete data

Visual: Data Distribution

KTable – Partitioned State (Sharded Across Instances)

Each Kafka Streams instance stores only a portion of the table, based on Kafka topic partitioning:

KTable: Product Table (partitioned across 3 instances)

Instance-1         Instance-2         Instance-3
─────────────      ─────────────      ─────────────
productId: 101     productId: 102     productId: 103
productId: 104     productId: 105     productId: 106
productId: 107     productId: 108     productId: 109

Each instance holds only the product records assigned to its partition.

Joins work only if the stream's key matches the table's key  
(and both are co-partitioned, using the same key - productId).

GlobalKTable – Replicated State (Full Copy in Every Instance)

Each Kafka Streams instance gets a full copy of the entire table — all partitions:

GlobalKTable: Product Table (replicated in every instance)

Instance-1         Instance-2         Instance-3
─────────────      ─────────────      ─────────────
productId: 101     productId: 101     productId: 101
productId: 102     productId: 102     productId: 102
productId: 103     productId: 103     productId: 103
productId: 104     productId: 104     productId: 104
productId: 105     productId: 105     productId: 105
productId: 106     productId: 106     productId: 106
productId: 107     productId: 107     productId: 107
productId: 108     productId: 108     productId: 108
productId: 109     productId: 109     productId: 109

Every instance has access to all keys and values in the table.

Ideal for lookups using foreign keys, where keys don't match  
the stream's partitioning.

Comparison: KTable vs GlobalKTable

┌─────────────────────────────────────────────────┐
│  KTABLE (Partitioned)                           │
├─────────────────────────────────────────────────┤
│                                                 │
│  Topic: "users" (4 partitions)                  │
│                                                 │
│  Instance 1:           Instance 2:              │
│  ├─ Partition 0        ├─ Partition 2          │
│  └─ Partition 1        └─ Partition 3          │
│                                                 │
│  Each sees 50% of data                          │
└─────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────┐
│  GLOBALKABLE (Replicated)                       │
├─────────────────────────────────────────────────┤
│                                                 │
│  Topic: "users" (4 partitions)                  │
│                                                 │
│  Instance 1:           Instance 2:              │
│  ├─ Partition 0        ├─ Partition 0          │
│  ├─ Partition 1        ├─ Partition 1          │
│  ├─ Partition 2        ├─ Partition 2          │
│  └─ Partition 3        └─ Partition 3          │
│                                                 │
│  Each sees 100% of data                         │
└─────────────────────────────────────────────────┘

GlobalKTable Characteristics

GLOBALKABLE PROPERTIES:

1. Full Replication
   └─ Every instance has complete copy
   └─ All partitions consumed

2. No Partitioning
   └─ Data not partitioned across instances
   └─ Every instance sees all keys

3. Read-Only
   └─ Used for lookups/enrichment
   └─ No writes to GlobalKTable

4. Higher Memory
   └─ Each instance stores full dataset
   └─ Trade memory for flexibility

Why GlobalKTable Exists

PROBLEM WITH KTABLE:
Stream: "orders" (partition by order_id)
Table:  "users"  (partition by user_id)

Join orders with users:
├─ Partitioning doesn't match!
├─ order_id ≠ user_id
└─ Requires expensive repartitioning

SOLUTION WITH GLOBALKABLE:
Stream: "orders" (partition by order_id)
GlobalKTable: "users" (fully replicated)

Join orders with users:
├─ No repartitioning needed!
├─ Every instance has all users
└─ Can join any order with any user

Use Cases for GlobalKTable

✅ Small reference data
   Example: Country codes, product catalog

✅ Foreign-key joins
   Example: Join orders (by order_id) with users (by user_id)

✅ Broadcast data
   Example: Configuration, feature flags

✅ When data small enough to fit in memory
   Example: < 100 MB per instance

❌ Do NOT use for:
   └─ Large datasets (> 1 GB)
   └─ High-frequency updates
   └─ When partitioned KTable suffices

⚠️ Caution: Memory Constraints with GlobalKTable

Critical Limitation: If a GlobalKTable is too large to fit into memory on each instance, your application will fail with OutOfMemoryError.

Memory Calculation:
├─ Each instance must hold: Full dataset + processing overhead
├─ Safe guideline: Keep GlobalKTable < 100 MB per instance
├─ Maximum recommended: < 1 GB (with sufficient heap)
└─ Beyond that: Use regular KTable (partitioned)

Example Problem:
├─ GlobalKTable size: 5 GB
├─ 10 instances × 5 GB each = 50 GB total memory
└─ Unacceptable! Use KTable instead.

When KTable is NOT Enough (Use GlobalKTable):

Use GlobalKTable when:

1. Partition alignment is not guaranteed or not possible:

   • Can't ensure stream and table use the same key or same partition count
   • Consuming from external topics with unknown or inconsistent partitioning
   • Joining against a small reference dataset (product catalog, country codes)

2. You need every record in the table available to every instance:

   • Like a lookup table: country → name, currency → symbol
   • Think of it as a replicated cache needed in full

3. You want to avoid complex repartitioning logic:

   • GlobalKTable avoids repartitioning entirely (every instance has full table)
   • Can join on any stream regardless of its key — even without .selectKey()

// Example: No repartitioning needed with GlobalKTable
ordersStream
    .join(globalProductTable,
        (order) -> order.getProductId(),  // Key mapper
        (order, product) -> enrich(order, product));

// vs with KTable (requires repartitioning):
ordersStream
    .selectKey((key, order) -> order.getProductId())  // Change key
    .through("orders-by-product")                      // Repartition
    .join(productTable, ...)                           // Then join

6. Comparison Table

7. Choosing the Right Abstraction

Decision Tree

Do you need ALL events or just LATEST per key?
├─ ALL events → KStream
└─ LATEST per key → KTable or GlobalKTable
                    │
                    └─ Is data small (<100MB)?
                       ├─ YES → Consider GlobalKTable
                       │        (for foreign-key joins)
                       └─ NO → Use KTable

Example Scenarios

Scenario 1: Track all user clicks
└─ Use: KStream
└─ Why: Every click is independent event

Scenario 2: Current user profile
└─ Use: KTable
└─ Why: Only latest profile matters

Scenario 3: Product catalog (small, for enrichment)
└─ Use: GlobalKTable
└─ Why: Need full catalog on every instance for joins

Scenario 4: Order history
└─ Use: KStream
└─ Why: Every order is important, not just latest

Scenario 5: Inventory levels
└─ Use: KTable
└─ Why: Only current level matters

Scenario 6: Country codes (for joins)
└─ Use: GlobalKTable
└─ Why: Small dataset, needed for foreign-key joins

8. Conceptual Code Examples

KStream Example

// Event stream: every click is preserved
KStream<String, ClickEvent> clicks = builder.stream("user-clicks");

clicks.foreach((userId, click) -> {
    System.out.println("User " + userId + " clicked " + click.getPage());
});

// Output (all events):
// User alice clicked home
// User bob clicked products
// User alice clicked cart
// User alice clicked checkout
// User bob clicked home

KTable Example

// Changelog: only latest value per key
KTable<String, UserProfile> users = builder.table("user-profiles");

users.toStream().foreach((userId, profile) -> {
    System.out.println("User " + userId + " has profile " + profile);
});

// Output (only latest per user):
// User alice has profile {name: Alice, age: 26}
// User bob has profile {name: Bob, age: 31}

GlobalKTable Example

// Fully replicated reference data
GlobalKTable<String, Country> countries = 
    builder.globalTable("countries");

// Available for lookups on any instance
// No partitioning constraints

9. Under the Hood (Conceptual)

KStream Implementation

KStream reads from Kafka topic:
├─ Processes records as they arrive
├─ No state stored (stateless by default)
├─ Records flow through processing pipeline
└─ Results written to output topic

KTable Implementation

KTable reads from Kafka topic:
├─ Maintains state store (RocksDB)
├─ Each new record updates state
├─ Latest value per key always available
├─ State backed up to changelog topic
└─ Can query current state

GlobalKTable Implementation

GlobalKTable reads from Kafka topic:
├─ Consumes ALL partitions (not just assigned ones)
├─ Maintains full state store on every instance
├─ Higher memory usage
├─ No coordination needed
└─ Useful for broadcast/reference data

10. Common Patterns

Pattern 1: Stream of Events → Table of Counts

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

KTable<String, Long> clickCounts = clicks
    .groupByKey()
    .count(); // Stream → Table aggregation

// Input (Stream): All click events
// Output (Table): Current count per user

Pattern 2: Table of Changes → Stream of Updates

KTable<String, UserProfile> users = builder.table("users");

KStream<String, UserProfile> userChanges = users.toStream();

// Input (Table): Current user profiles
// Output (Stream): All profile update events

Pattern 3: Enrich Stream with GlobalKTable

KStream<String, Order> orders = builder.stream("orders");
GlobalKTable<String, Product> products = builder.globalTable("products");

KStream<String, EnrichedOrder> enrichedOrders = orders.join(
    products,
    (orderId, order) -> order.getProductId(), // Foreign key
    (order, product) -> enrich(order, product) // Enrichment logic
);

// Join any order with any product (no repartitioning)

11. Mental Model Summary

The Core Abstractions

KSTREAM = Log of events
├─ "This happened, then this, then this..."
├─ All events preserved
└─ Insert semantics

KTABLE = Current state
├─ "This is the current value"
├─ Only latest per key
└─ Update semantics

GLOBALKABLE = Replicated state
├─ "Everyone has full copy"
├─ All instances see all data
└─ For reference/lookups

When to Use Each

Use KStream when:
✅ Every event matters
✅ Building audit log
✅ Time-series analysis
✅ Event-driven workflows

Use KTable when:
✅ Only current value matters
✅ Need latest state per key
✅ Aggregating streams
✅ Maintaining materialized view

Use GlobalKTable when:
✅ Small reference data
✅ Foreign-key joins needed
✅ Broadcast data to all instances
✅ Data fits in memory

12. Questions You Should Be Able to Answer

After this module, you should be able to answer:

✅ What is the difference between KStream and KTable?
✅ What does "update semantics" mean in KTable?
✅ What is a tombstone record?
✅ When should I use GlobalKTable instead of KTable?
✅ Can I convert a KStream to KTable? How?
✅ What does "stream-table duality" mean?
✅ What are the memory implications of GlobalKTable?
✅ When should I use each abstraction?

13. What's Next?

Now that you understand the core abstractions, the next modules will teach you how to use them:

• Module 3.1: Building a Streams application — putting it all together
• Module 4.1: Stateless operations — transforming KStreams
• Module 5: Stateful operations — aggregating into KTables
• Module 8.1: Joins — combining streams and tables

Summary

You learned:

• ✅ KStream represents an event stream (all events preserved, immutable)
• ✅ KTable represents a changelog (latest value per key, mutable)
• ✅ GlobalKTable is a fully replicated KTable (all data on every instance)
• ✅ Insert vs update semantics determine which abstraction to use
• ✅ Stream-table duality — they are two views of the same data
• ✅ Use cases: events → KStream, current state → KTable, reference data → GlobalKTable
• ✅ Trade-offs: memory, partitioning, and access patterns differ

Next up: Module 3.1 — Building a Kafka Streams Application — Learn how to actually create a Streams app with code.