Stateful Stream Processing

​

What is state?

• Pattern detection: storing the sequence of events seen so far while searching for a complex event pattern
• Windowed aggregations: maintaining running counts, sums, or averages within a time window
• Machine learning: holding current model parameters while training over a stream of data points
• Deduplication: keeping a set of already-seen IDs to filter out duplicate events
• Session tracking: maintaining per-user session data across a series of interactions

1. Make it fault tolerant via checkpoints and savepoints
2. Enable rescaling — redistributing state across parallel instances when you change parallelism

​

Keyed state

 Stream of (userId, action) events

     keyBy(userId)
         │
    ┌────┴──────────────────────────────┐
    │  Keys A–M    │  Keys N–Z          │
    │  ┌─────────┐ │  ┌─────────┐      │
    │  │userId=A │ │  │userId=P │      │
    │  │  state  │ │  │  state  │      │
    │  └─────────┘ │  └─────────┘      │
    └──────────────┴────────────────────┘
      Subtask 1         Subtask 2

​

Key groups

​

Keyed state types

// Declare state in open()
ValueStateDescriptor<Long> descriptor =
    new ValueStateDescriptor<>("last-seen", Long.class);
ValueState<Long> lastSeen = getRuntimeContext().getState(descriptor);

// Use in processElement()
Long previous = lastSeen.value(); // null if key is new
lastSeen.update(currentTimestamp);

ListStateDescriptor<String> descriptor =
    new ListStateDescriptor<>("events", String.class);
ListState<String> eventList = getRuntimeContext().getListState(descriptor);

// Append to the list
eventList.add(event.getPayload());

// Iterate over all stored values
for (String e : eventList.get()) {
    // process each stored event
}

MapStateDescriptor<String, Integer> descriptor =
    new MapStateDescriptor<>("feature-counts", String.class, Integer.class);
MapState<String, Integer> featureCounts = getRuntimeContext().getMapState(descriptor);

// Increment count for a feature
String feature = event.getFeatureName();
int current = featureCounts.contains(feature)
    ? featureCounts.get(feature) : 0;
featureCounts.put(feature, current + 1);

// ReducingState example: sum of longs
ReducingStateDescriptor<Long> descriptor =
    new ReducingStateDescriptor<>("sum", Long::sum, Long.class);
ReducingState<Long> sum = getRuntimeContext().getReducingState(descriptor);

sum.add(event.getValue());    // merges immediately
Long currentSum = sum.get();  // retrieve result

​

Operator state

• Even split: the list of state items is split evenly across the new parallel instances
• Union: every new parallel instance receives a full copy of the union of all state

​

State backends

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new HashMapStateBackend());

env.setStateBackend(new EmbeddedRocksDBStateBackend());
// Enable incremental checkpoints
env.setStateBackend(new EmbeddedRocksDBStateBackend(true));

​

State persistence and checkpointing

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// Enable checkpointing every 60 seconds
env.enableCheckpointing(60_000);

​

Checkpoint barriers

  Source stream:
  ... [record] [record] [barrier n] [record] [record] [barrier n+1] ...
                           │
                    Separates snapshot n
                    from snapshot n+1

1. Takes a snapshot of its current state (asynchronously)
2. Emits barrier n downstream
3. Continues processing records

​

Aligned vs. unaligned checkpoints

• When barrier n arrives from input stream A but not yet from stream B, the operator buffers records from stream A
• Once barrier n arrives from all inputs, the operator snapshots and forwards the barrier
• This guarantees exactly-once but adds latency for slow or backpressured inputs

• The operator immediately forwards barrier n downstream without waiting for other inputs
• All in-flight records (those that overtook the barrier) are included in the checkpoint state
• Lower checkpoint latency for pipelines with persistent backpressure, but higher I/O overhead

env.enableCheckpointing(60_000);
env.getCheckpointConfig().enableUnalignedCheckpoints();

​

Exactly-once vs. at-least-once

import org.apache.flink.streaming.api.CheckpointingMode;

env.enableCheckpointing(60_000);
env.getCheckpointConfig().setCheckpointingConsistencyMode(
    CheckpointingMode.EXACTLY_ONCE  // default
    // or CheckpointingMode.AT_LEAST_ONCE for lower latency
);

​

Recovery

1. Flink selects the latest completed checkpoint k
2. It re-deploys the entire distributed dataflow
3. Each operator is restored to the state it had at checkpoint k
4. Sources start reading from position S_k (e.g., a Kafka partition offset)

​

Savepoints

# Trigger a savepoint for a running job
./bin/flink savepoint <jobId> [targetDirectory]

# Restore a job from a savepoint
./bin/flink run \
  --fromSavepoint /path/to/savepoint \
  ./my-flink-job.jar

• Upgrade Flink to a new version without losing state
• Update job logic while preserving accumulated state
• Change parallelism — state is redistributed across the new number of parallel instances
• Clone a job — start two versions of the same job from the same state for A/B testing

DataStream<String> stream = env
    .addSource(new KafkaSource<>(...)).uid("kafka-source")
    .map(new MyMapper()).uid("my-mapper")
    .keyBy(record -> record.getUserId());

​

State in batch programs

• Batch programs do not use checkpointing for fault tolerance. Recovery happens by fully replaying the bounded input.
• State backends in batch execution mode use simplified in-memory or out-of-core data structures rather than full key/value indexes.