Chapter 36: Stream Processing

What is Stream Processing

Stream processing operates on unbounded data sets—continuous flows of events arriving over time with no predetermined end. Unlike batch processing that waits for all data before computing results, stream processing produces continuous outputs with low latency as events arrive.

Unbounded vs Bounded Data

Bounded (batch) data has a fixed size—process everything, then finish. Unbounded (streaming) data keeps arriving indefinitely. Stream processors handle infinite inputs by computing incremental results using windows and triggers.

Event Time vs Processing Time

• Event Time: When the event actually occurred (timestamp in data). Correct for out-of-order data.
• Processing Time: When the system processes the event. Simplest but incorrect if data arrives late.
• Ingestion Time: When event enters the stream processor. Hybrid approach.

Latency Requirements

Stream processing targets sub-second to few-second latency. Fraud detection needs milliseconds. Real-time dashboards tolerate seconds. If latency requirements exceed minutes, batch processing may be more appropriate.

Core Concepts

Events and Streams

Event: An immutable fact that happened at a point in time (user clicked, sensor reading, transaction).

Stream: An ordered sequence of events. Streams are partitioned for parallelism—events with the same key go to the same partition to maintain ordering.

Operators

Operators transform streams: map (1-to-1), filter (drop events), flatMap (1-to-many), keyBy (partition by key), reduce/aggregate (combine), join (combine two streams), window (group by time).

State

State is information remembered across events. Stateless operators process each event independently. Stateful operators maintain context—aggregations, joins, and windowing require state. State must be fault-tolerant via checkpointing.

Time

Time determines when to trigger computations. Event time uses timestamps in data. Processing time uses wall-clock time. Watermarks track progress in event time.

Windows

Windows divide infinite streams into finite chunks for computation. Without windows, aggregations would never produce output.

Conceptual stream processing pipeline with stateful windowing operator

Windowing

Windows assign events to finite groups for aggregation. Window types define how events are grouped.

Tumbling Windows

Fixed-size, non-overlapping windows. Events belong to exactly one window. Use for periodic aggregations like "total orders per minute."

Sliding Windows

Fixed-size, overlapping windows. Events belong to multiple windows. Use for moving averages like "average latency over last 5 minutes, updated every minute."

Session Windows

Dynamic-size windows based on inactivity gaps. Window closes after a timeout with no events. Use for user sessions—"aggregate clicks until 30 minutes of inactivity."

Global Windows

Single window containing all events. Requires custom triggers. Use for custom windowing logic not covered by standard types.

Time & Watermarks

The Late Data Problem

In event-time processing, events arrive out of order due to network delays, clock skew, or backfilling. When should we close a window and emit results if late data might still arrive?

Watermarks

A watermark is an assertion: "No events with timestamp less than T will arrive." When a watermark passes the end of a window, the window closes and emits results. Watermarks are heuristics—they balance latency (close windows quickly) vs completeness (wait for late data).

Perfect Watermark: Never early—waits for all data. High latency, perfect accuracy.

Heuristic Watermark: Based on observed delays (e.g., "99th percentile is 10 seconds"). Low latency, occasional late data.

Allowed Lateness

After a watermark closes a window, allow a grace period for late data. Late events update the already-emitted result. After the grace period, discard or side-output late events.

Side Outputs

Late events after allowed lateness are sent to a side output stream for manual inspection, logging, or reprocessing.

Window closes at watermark, accepts late events during grace period, drops events after

stream
  .assignTimestampsAndWatermarks(
    WatermarkStrategy
      .forBoundedOutOfOrderness(Duration.ofSeconds(10))  // Max 10s out-of-order
      .withTimestampAssigner((event, timestamp) -> event.getTimestamp())
  )
  .keyBy(event -> event.getUserId())
  .window(TumblingEventTimeWindows.of(Time.minutes(5)))
  .allowedLateness(Time.seconds(30))  // Grace period
  .sideOutputLateData(lateDataTag)
  .aggregate(new CountAggregator());

Apache Flink

Flink is a distributed stream processing framework with true streaming (not micro-batches), exactly-once semantics, low latency, and powerful state management.

Architecture

JobManager: Coordinates the job—schedules tasks, manages checkpoints, handles failures. Highly available via Zookeeper/Kubernetes.

TaskManager: Worker nodes that execute tasks. Each TaskManager has task slots for parallel execution.

Dataflow Graphs

Jobs are directed acyclic graphs (DAGs) of operators. Source operators read from Kafka/files. Transformation operators apply logic. Sink operators write to databases/Kafka. Flink optimizes the graph (operator chaining, pipelining).

Exactly-Once Semantics

Flink achieves exactly-once via distributed snapshots (checkpointing). Periodic checkpoints capture consistent state across all operators. On failure, restore from checkpoint and replay from source (requires source support like Kafka offsets).

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(60000);  // Checkpoint every 60s

DataStream<Event> stream = env
  .addSource(new FlinkKafkaConsumer<>("topic", new EventSchema(), props))
  .assignTimestampsAndWatermarks(...)
  .keyBy(event -> event.getUserId())
  .window(TumblingEventTimeWindows.of(Time.minutes(5)))
  .aggregate(new SumAggregator())
  .addSink(new JdbcSink(...));

env.execute("Flink Job");

Parallelism

Each operator has a parallelism (number of instances). Data is partitioned across instances. keyBy ensures events with same key go to same instance (required for stateful operations). Operators with different parallelism are connected via shuffle/rebalance.

State Management

State enables operators to remember information across events. Flink manages state with fault tolerance via checkpointing.

Keyed State

Scoped to a specific key (after keyBy). Each key has its own state. Types:

• ValueState<T>: Single value per key (last seen value)
• ListState<T>: List of values per key (accumulate events)
• MapState<K,V>: Key-value map per key (nested structure)
• ReducingState<T>: Single value reduced via ReduceFunction
• AggregatingState<IN,OUT>: Single value aggregated via AggregateFunction

Operator State

Scoped to an operator instance (not keyed). Types:

• ListState: List redistributed across instances on rescaling
• UnionListState: Full list broadcast to all instances on rescaling
• BroadcastState: Read-only state broadcast to all instances

State Backends

State backends determine how state is stored and checkpointed.

RocksDB: Embedded key-value store. Stores state on disk, spills to disk when exceeds memory. Supports incremental checkpoints (only checkpoint changed data). Default for production.

State TTL

Automatically expires state after a TTL to prevent unbounded growth. Configure update strategy (on create/read/write) and expiration (on access/processing time).

public class CountFunction extends KeyedProcessFunction<String, Event, Result> {
  private transient ValueState<Long> countState;

  @Override
  public void open(Configuration config) {
    StateTtlConfig ttlConfig = StateTtlConfig
      .newBuilder(Time.days(7))
      .setUpdateType(UpdateType.OnCreateAndWrite)
      .build();

    ValueStateDescriptor<Long> descriptor =
      new ValueStateDescriptor<>("count", Long.class);
    descriptor.enableTimeToLive(ttlConfig);

    countState = getRuntimeContext().getState(descriptor);
  }

  @Override
  public void processElement(Event event, Context ctx, Collector<Result> out) {
    Long count = countState.value();
    count = (count == null) ? 1 : count + 1;
    countState.update(count);
    out.collect(new Result(ctx.getCurrentKey(), count));
  }
}

Queryable State

Expose state for external queries without affecting processing. Useful for debugging or serving low-latency lookups from stream state.

Checkpointing

Checkpointing creates consistent snapshots of state across all operators for fault tolerance. On failure, restore from checkpoint and replay from source.

Chandy-Lamport Algorithm

Flink uses a variant of Chandy-Lamport for distributed snapshots. JobManager injects checkpoint barriers into sources. Barriers flow through the dataflow graph. When an operator receives barriers from all inputs, it checkpoints its state and forwards barriers downstream.

Barrier Alignment

Operators with multiple inputs wait for barriers from all inputs before checkpointing (alignment). This ensures consistency but increases latency.

Aligned Checkpoints: Wait for barriers from all inputs. Exactly-once semantics. Higher latency during checkpoint.

Unaligned Checkpoints: Don't wait—checkpoint in-flight data in buffers. Exactly-once semantics. Lower latency, larger checkpoints.

Exactly-Once vs At-Least-Once

Exactly-Once: Each event affects state exactly once. Requires checkpointing and transactional sinks (two-phase commit). Higher overhead.

At-Least-Once: Events may be reprocessed on failure. Simpler, lower overhead, but requires idempotent operations.

Incremental Checkpoints

With RocksDB, only checkpoint state changes since last checkpoint (not full state). Dramatically reduces checkpoint size and time for large state. Increases recovery time (must apply incremental changes).

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Enable checkpointing every 60 seconds
env.enableCheckpointing(60000);

// Checkpoint config
CheckpointConfig config = env.getCheckpointConfig();
config.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
config.setMinPauseBetweenCheckpoints(30000);  // Min 30s between checkpoints
config.setCheckpointTimeout(600000);  // Timeout after 10 min
config.setMaxConcurrentCheckpoints(1);
config.enableUnalignedCheckpoints();  // Lower latency

// Retain checkpoints on cancellation for manual recovery
config.setExternalizedCheckpointCleanup(
  ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

// RocksDB with incremental checkpoints
env.setStateBackend(new EmbeddedRocksDBStateBackend(true));  // true = incremental
env.getCheckpointConfig().setCheckpointStorage("s3://bucket/checkpoints");

Kafka Streams

Kafka Streams is a lightweight stream processing library (not a separate cluster). Applications are standard Java/Scala programs that read/write Kafka topics.

Architecture

No separate cluster—deploy as microservices. Scales horizontally by adding instances. Leverages Kafka's partitioning for parallelism and consumer groups for coordination. State stored locally (RocksDB) and backed up to Kafka changelog topics.

KStream and KTable

KStream: Unbounded stream of events. Each record is an insert. Represents event log.

KTable: Changelog stream—latest value per key. Each record is an upsert. Represents materialized view.

GlobalKTable: Fully replicated KTable (not partitioned). All instances have full table for joins.

Operations

• Stateless: map, filter, flatMap, branch, merge
• Stateful: aggregate, reduce, count, join, windowing
• Joins: KStream-KStream (windowed), KStream-KTable (lookup), KTable-KTable (upsert)

Interactive Queries

Query local state stores via REST API. Discover which instance holds a key (metadata), route query to that instance. Enables serving layer from stream state.

StreamsBuilder builder = new StreamsBuilder();

KStream<String, Event> events = builder.stream("events-topic");

KTable<String, Long> counts = events
  .groupByKey()
  .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
  .count(Materialized.as("counts-store"));  // Local RocksDB store

counts.toStream().to("counts-topic");

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

Flink vs Kafka Streams

Spark Structured Streaming

Structured Streaming is Spark's stream processing API built on Spark SQL. Treats streams as unbounded tables with incremental queries.

Micro-Batch vs Continuous Mode

Micro-Batch (default): Process data in small batches (100ms-1s). Higher latency but higher throughput. Leverages Spark's batch optimization.

Continuous Mode (experimental): True streaming with ~1ms latency. Limited operators. Use for ultra-low latency.

Watermarks and Late Data

Define watermarks via withWatermark(). Late data within watermark delay is processed. Data beyond watermark is dropped.

val events = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "events")
  .load()

val counts = events
  .selectExpr("CAST(value AS STRING) as json")
  .select(from_json($"json", schema).as("data"))
  .select("data.*")
  .withWatermark("timestamp", "10 minutes")  // 10 min late data allowed
  .groupBy(window($"timestamp", "5 minutes"), $"userId")
  .count()

val query = counts
  .writeStream
  .format("console")
  .outputMode("update")  // Only updated rows
  .start()

query.awaitTermination()

Output Modes

Append: Only new rows since last trigger. Use for non-aggregations or windowed aggregations after watermark.

Update: Only updated rows since last trigger. Use for aggregations.

Complete: Entire result table. Use for small aggregations (in-memory).

State Management

State automatically managed for aggregations and joins. State stored in memory/disk. Checkpointing to HDFS/S3 for fault tolerance. mapGroupsWithState and flatMapGroupsWithState for custom state.

Design Patterns

Event-Driven Microservices

Microservices communicate via event streams (Kafka). Each service consumes events, processes, produces events. Services are decoupled—schema evolution via Avro/Protobuf. Use Kafka Streams or Flink for stateful processing within services.

Real-Time ETL

Transform and enrich events in real-time before storing. Read from source (Kafka, Kinesis), apply transformations (filter, map, join with lookup tables), write to sink (data lake, warehouse). Reduces batch ETL latency from hours to seconds.

Fraud Detection

Detect fraudulent transactions in real-time. Use CEP (complex event processing) to match patterns (e.g., "3 failed logins followed by successful login from different country"). Use stateful operators to track user behavior (velocity checks, spending patterns). Low latency critical—block transaction before completion.

// Flink CEP example
Pattern<Event, ?> pattern = Pattern.<Event>begin("first")
  .where(evt -> evt.getType().equals("LOGIN_FAILED"))
  .times(3).consecutive()
  .within(Time.minutes(5))
  .next("second")
  .where(evt -> evt.getType().equals("LOGIN_SUCCESS") &&
                evt.getCountry() != evt.getPreviousCountry());

PatternStream<Event> patternStream = CEP.pattern(eventStream, pattern);

DataStream<Alert> alerts = patternStream.select(new PatternSelectFunction<>() {
  public Alert select(Map<String, List<Event>> pattern) {
    return new Alert("Potential account takeover", pattern.get("second").get(0));
  }
});

Sessionization

Group user events into sessions using session windows. Session closes after inactivity timeout. Use for user analytics—session duration, pages per session, conversion tracking. State per user key tracks session.

Real-Time Dashboards

Aggregate events and push to dashboard. Use tumbling windows for periodic updates (every minute). Windowed aggregations emit to sink (Elasticsearch, TimescaleDB). Frontend polls or uses WebSockets for live updates. Latency typically 1-10 seconds.

Complex Event Processing (CEP)

Match sequences of events over time. Use Flink CEP library to define patterns (sequence, followed by, or, not, within time). Detect business-critical patterns—SLA violations, system failures, user journeys. Stateful pattern matching across event streams.

Framework Comparison

Decision Framework

• Use Flink if: Sub-second latency required, large state (TBs), complex CEP, exactly-once critical, multi-source processing
• Use Kafka Streams if: Kafka-only sources/sinks, microservice architecture, prefer library over cluster, moderate state (<100GB)
• Use Spark Structured Streaming if: Unified batch/stream codebase, existing Spark ecosystem, latency >100ms acceptable, leverage Spark SQL
• Use Managed (Kinesis/Dataflow) if: Minimize operational overhead, cloud-native, standard processing logic

Key Tradeoffs

Latency vs Throughput: Micro-batching trades latency for throughput. True streaming has lower latency but requires more careful tuning.

Exactly-Once vs Performance: Exactly-once semantics add overhead (checkpointing, two-phase commit). At-least-once is faster but requires idempotent operations.

State Size vs Operational Complexity: Large state (TBs) requires careful backend selection (RocksDB), incremental checkpoints, and operational expertise. Small state simplifies operations.

Deployment Model: Separate cluster (Flink) provides isolation and scalability but adds operational complexity. Library (Kafka Streams) simplifies deployment but couples processing to application lifecycle. Managed services minimize operations but reduce control.