Kafka Streams: DSL, Windowed Aggregation, and Exactly-Once

Kafka Streams: Real-Time Stream Processing with the DSL

Kafka Streams is a client library for building real-time stream processing applications that consume from and produce to Kafka topics.

Why Kafka Streams?

Key Advantage:

• No separate cluster required
• Runs as an embedded library within your application
• Leverages Kafka's own partitioning for parallelism and fault tolerance

Capabilities:

1. Exactly-once semantics — reliable processing
2. Stateful aggregations — maintain state across events
3. Windowed operations — time-based grouping
4. Interactive queries — query state stores directly

Key Insight: For organizations already invested in Kafka, Streams eliminates the operational overhead of maintaining Flink or Spark clusters.

Architecture Diagram

Key Concepts

Production Code

Windowed Session Aggregation

from confluent_kafka import Producer
from kafka import KafkaConsumer
from kafka.streams import StreamsBuilder, KStream, KTable, TimeWindows, SessionWindows
from kafka.streams.kstream import Materialized, Grouped, Consumed, Produced
import json
import logging

logger = logging.getLogger(__name__)

# This example uses Java Kafka Streams via Py4J bridge
# For production Python Kafka Streams, use faust or bytewax
# This demonstrates the conceptual topology

def build_session_aggregation_topology():
    """Build a Kafka Streams topology for session-based click aggregation."""
    from py4j.java_gateway import JavaGateway

    gateway = JavaGateway()
    streams_builder = gateway.entry_point.StreamsBuilder()

    # Source: click events from Kafka topic
    click_stream = streams_builder.stream(
        "click-events",
        gateway.entry_point.Consumed.with(
            gateway.entry_point.Serdes.String(),
            gateway.entry_point.Serdes.String(),
        ),
    )

    # Rekey by user_id for co-partitioned aggregation
    keyed_stream = click_stream.select_key(
        lambda key, value: json.loads(value)["user_id"]
    )

    # Group by user_id
    grouped = keyed_stream.group_by_key(
        gateway.entry_point.Grouped.with(
            gateway.entry_point.Serdes.String(),
            gateway.entry_point.Serdes.String(),
        )
    )

    # Session window: 30-min inactivity gap, 24-hour max duration
    session_window = gateway.entry_point.SessionWindows.of_with_no_grace(
        gateway.entry_point.Duration.of_minutes(30)
    )

    # Aggregate clicks per session
    session_counts = grouped.windowed_by(
        session_window
    ).count(
        "session-click-counts",
        gateway.entry_point.Materialized.as(
            "session-click-counts-store"
        ).with_key_serde(gateway.entry_point.Serdes.String())
         .with_value_serde(gateway.entry_point.Serdes.Long())
    )

    # Convert to stream and produce to output topic
    session_counts.to_stream().to(
        "session-click-aggregations",
        gateway.entry_point.Produced.with(
            gateway.entry_point.Serdes.String(),
            gateway.entry_point.Serdes.Long(),
        ),
    )

    topology = streams_builder.build()
    logger.info(f"Topology:\n{topology.describe()}")
    return topology


# Python-native approach using bytewax (Kafka Streams equivalent)
from bytewax.connectors.kafka import KafkaSource, KafkaSink
from bytewax.dataflow import Dataflow
from bytewax.window import SessionWindow, TumblingWindow, SystemTimeAssigner
from bytewax.operators import max_all, min_all
import json
from datetime import timedelta


def build_bytewax_session_aggregation():
    """Build a bytewax Dataflow for session-based aggregation (Python-native)."""
    from bytewax.inputs import KafkaSourceConfig
    from bytewax.outputs import KafkaSinkConfig

    # Define the Dataflow
    flow = Dataflow("click-session-aggregation")

    # Kafka source: consume click events
    flow.input(
        "kafka-in",
        KafkaSourceConfig(
            brokers=["kafka-broker-1:9092"],
            topic="click-events",
            group_id="session-aggregation",
        ),
    )

    # Parse JSON and extract user_id
    flow.map("parse", lambda msg: json.loads(msg.value))
    flow.key_on("key-by-user", lambda event: event["user_id"])

    # Session window with 30-min gap
    session_win = SessionWindow(gap=timedelta(minutes=30))
    flow.window("session-window", session_win, assigner=SystemTimeAssigner())

    # Count events per session
    flow.count("count-per-session")

    # Kafka sink: produce aggregated results
    flow.output(
        "kafka-out",
        KafkaSinkConfig(
            brokers=["kafka-broker-1:9092"],
            topic="session-click-aggregations",
        ),
    )

    return flow

Stream-Table Join with Stateful Enrichment

import json
import logging
from typing import Dict, Optional
from dataclasses import dataclass, asdict

logger = logging.getLogger(__name__)


@dataclass
class OrderEvent:
    order_id: str
    customer_id: str
    product_id: str
    amount: float
    timestamp: str


@dataclass
class CustomerProfile:
    customer_id: str
    name: str
    tier: str
    credit_limit: float


class StreamTableJoinProcessor:
    """
    Simulates a Kafka Streams stream-table join for order enrichment.
    
    In Kafka Streams, this would be implemented as:
        KStream<order_id, OrderEvent> orders = ...;
        KTable<customer_id, CustomerProfile> customers = ...;
        KStream<order_id, EnrichedOrder> enriched = orders.join(
            customers,
            (order, profile) -> enrich(order, profile),
            Joined.with(orderKeySerde, orderSerde, profileSerde)
        );
    
    Parameters:
        None (state is managed internally)
        
    State:
        customer_table (Dict): Materialized KTable of customer profiles
        order_stream (List): Processed order events (for demonstration)
    """

    def __init__(self):
        self.customer_table: Dict[str, CustomerProfile] = {}
        self.order_stream: list = []

    def process_customer_update(self, record: dict) -> None:
        """
        Process customer profile updates into the materialized table.
        
        In Kafka Streams, this happens automatically when consuming from
        a customer-profiles topic with KTable.
        
        Parameters:
            record (dict): Customer profile update event
        """
        profile = CustomerProfile(**record)
        self.customer_table[profile.customer_id] = profile
        logger.debug(f"Updated customer table: {profile.customer_id}")

    def process_order_event(self, event: dict) -> Optional[dict]:
        """
        Join order event with customer table for enrichment.
        
        Parameters:
            event (dict): Order event from Kafka stream
            
        Returns:
            dict: Enriched order with customer profile data
        """
        order = OrderEvent(**event)
        customer = self.customer_table.get(order.customer_id)

        if customer is None:
            logger.warning(
                f"No customer profile for {order.customer_id}, "
                f"order {order.order_id} will be processed without enrichment"
            )
            enriched = {
                **asdict(order),
                "customer_name": "UNKNOWN",
                "tier": "unknown",
                "credit_limit": 0.0,
                "within_credit_limit": False,
            }
        else:
            enriched = {
                **asdict(order),
                "customer_name": customer.name,
                "tier": customer.tier,
                "credit_limit": customer.credit_limit,
                "within_credit_limit": order.amount <= customer.credit_limit,
            }

        self.order_stream.append(enriched)
        return enriched

    def get_table_stats(self) -> dict:
        """
        Return statistics about the materialized customer table.
        
        Returns:
            dict: Total customer count and tier distribution
        """
        return {
            "total_customers": len(self.customer_table),
            "tiers": {
                tier: sum(1 for p in self.customer_table.values() if p.tier == tier)
                for tier in set(p.tier for p in self.customer_table.values())
            }
        }


# Example usage simulating stream-table join
processor = StreamTableJoinProcessor()

# Simulate customer profile updates (table side)
customer_updates = [
    {"customer_id": "C001", "name": "Alice", "tier": "gold", "credit_limit": 10000.0},
    {"customer_id": "C002", "name": "Bob", "tier": "silver", "credit_limit": 5000.0},
    {"customer_id": "C003", "name": "Charlie", "tier": "bronze", "credit_limit": 1000.0},
]

for update in customer_updates:
    processor.process_customer_update(update)

# Simulate order events (stream side)
order_events = [
    {"order_id": "ORD-1", "customer_id": "C001", "product_id": "P100", "amount": 299.99, "timestamp": "2024-01-15T10:00:00Z"},
    {"order_id": "ORD-2", "customer_id": "C002", "product_id": "P200", "amount": 750.00, "timestamp": "2024-01-15T10:01:00Z"},
    {"order_id": "ORD-3", "customer_id": "C004", "product_id": "P300", "amount": 50.00, "timestamp": "2024-01-15T10:02:00Z"},  # Unknown customer
]

for event in order_events:
    result = processor.process_order_event(event)
    print(json.dumps(result, indent=2))

print(f"\nTable stats: {processor.get_table_stats()}")

Best Practices

1. Set processing.guarantee=exactly_once_v2 for production workloads that require exactly-once semantics.
2. Configure num.standby.replicas=1 to maintain warm standby state stores and reduce recovery time after failures.
3. Monitor state store size via JMX metrics. Alert on unexpected growth that indicates memory pressure.
4. Use TopologyTestDriver to unit test stream processing topologies without requiring a running Kafka cluster.
5. Set grace periods on windowed operations to handle late-arriving events within bounded timeframes.
6. Co-partition input topics for joins — both sides of a join must have the same number of partitions.
7. Use Suppress.untilTimeLimit() to buffer window updates and emit only the final result.
8. Monitor changelog lag to detect state store recovery delays after broker failures.
9. Set max.task.idle.ms to handle data skew in partitions — idle tasks will wait for straggler partitions.
10. Version your state stores using StoreBuilder.withVersion() to support rolling upgrades without state migration.

Kafka Streams vs Alternatives

See Also

• 019 - Apache Kafka: Topics, Producers, and Consumers - Kafka fundamentals and topic design
• 022 - Spark Structured Streaming - Alternative streaming engine
• 023 - Batch vs Streaming: Lambda, Kappa, and Architecture Trade-offs - Architecture pattern comparison
• 020 - Kafka Streams - This lesson
• 030 - Capstone Project: Real-Time Streaming Pipeline - Applying Kafka Streams in production