FsStateBackend may be the most commonly used StateBackend at many companies.

Characteristics and Uses

FsStateBackend keeps state in memory. State operations directly manipulate in-memory objects, avoiding disk and serialization costs. In one word: fast. Such a blunt design inevitably has drawbacks:

• State cannot exceed the JVM heap limit; a sudden traffic spike can immediately cause an OOM exit
• GC is a headache: continually creating and destroying state objects can make a job practically unusable

In production, moderately complex workloads such as joins show pronounced GC issues when TaskManager memory exceeds 10 GB. When is this backend suitable?

My general recommendation is to use it whenever state does not store raw records: for example, counting article reads every ten minutes or using bitmaps to calculate ten-minute unique visitors. In typical applications, the number of keys, such as articles or users, is bounded and unlikely to grow by orders of magnitude. Raw-record state is different: bad upstream data, skew, or abusive users can cause a single key’s state to grow enormously.

State Representation

First, a diagram:

Flink uses an array plus linked lists, similar to Java 1.7’s HashMap. StateMap exposes these interfaces:

1
public abstract S get(K key, N namespace);
2

3
public abstract boolean containsKey(K key, N namespace);
4

5
public abstract void put(K key, N namespace, S state);
6

7
public abstract S putAndGetOld(K key, N namespace, S state);
8

9
public abstract void remove(K key, N namespace);
10

11
public abstract S removeAndGetOld(K key, N namespace);

StateMap uses Key and Namespace together as the map key and state as its value. Each task’s StateBackend uses KeyGroup as a first-level index.

Snapshot Mechanism

FsStateBackend supports synchronous and asynchronous snapshots. A synchronous snapshot writes all StateMaps under all KeyGroups to HDFS when a checkpoint is triggered.

Asynchronous snapshots use copy-on-write to persist StateMap contents to HDFS consistently. They still have synchronous and asynchronous phases. The synchronous phase performs two main operations:

1. Snapshot the StateMapEntry[] array. This copies only object references and is therefore lightweight.
2. Increment globalVersion so later operations can identify entries copied after the snapshot.

Using the earlier diagram, a checkpoint synchronously increments globalVersion from 1 to 2, then starts writing StateMap to HDFS asynchronously. If Entry5 must be updated during that write, the process is:

Every entry traversed while locating Entry5 is copied. Since the asynchronous snapshot already captured the object references in StateMapEntry[], this does not affect the ongoing HDFS write.

Other Details

Restoration essentially rereads the HDFS files to rebuild StateMap. Given the backend’s clear limitations, the community proposed improvements [1]. The idea is simple: some KeyGroups contain hot data, others cold data; when memory fills, persist cold data to disk. The diagram shows this:

Two components are introduced: HeapStatusMonitor and Spill / Load Manager. Specifically:

• HeapStatusMonitor: Since in-memory object sizes cannot be directly estimated, it uses approximations:
  • Check heap usage
  • Check GC pauses
• Spill / Load Manager:
  • Track each KeyGroup’s size and request rate
  • Spill/load StateMap data at KeyGroup granularity

I think the main difficulty is choosing the granularity for distinguishing hot and cold data. KeyGroup may still be too coarse with the default 65536 groups, although maxParallelism can adjust persistence granularity. Overall, SpillableHeapKeyedStateBackend replaces HeapKeyedStateBackend and can be configured to behave identically.

References

1. FLIP-50: Spillable Heap State Backend