How Can Flink State Be Initialized Without a Checkpoint?

This requirement is common during Flink development and product iteration:

• State must be loaded when a Flink program starts for the first time.
• Changes to state-related classes prevent existing state from being reloaded.
• A checkpoint is corrupted and cannot be deserialized.

This is clearly essential in production; without it, future development and iteration are heavily constrained. I have explored several approaches from different angles and record them here. See the community email discussion here.

ConnectedStream

ConnectedStream is a clever workaround requiring no changes to Flink itself.

1
val recoverStream = env.addSource(sourceA)
2
val dataStream = env.addSource(sourceB)
3
recoverStream.connect(dataStream)

Connect the restoration stream to the normal stream. Subsequent aggregations must handle pre-aggregated data and raw records separately, for example:

1
override def add(value: IN, accumulator: ACC): ACC = {
2
    value match {
3
        case AGG(x) => accumulator.merege(value)
4
        case RECORD(x) => accumulator.add(value)
5
        case _ => accumulator
6
    }
7
}

This merges the two data types directly through Flink’s ConnectedStream. It is simple and workable, though not particularly elegant :), and has these consequences:

1. An additional stream introduces tasks matching its parallelism, yet serves no purpose after state restoration.
2. Later ETL stages must distinguish and handle the data types separately, which becomes cumbersome in a long pipeline.

Bravo

Bravo is an open-source project for Flink state.

Bravo is a convenient state reader and writer library leveraging the Flink’s batch processing capabilities. It supports processing and writing Flink streaming snapshots. At the moment it only supports processing RocksDB snapshots but this can be extended in the future for other state backends.

It reads and writes Flink savepoint directories, currently supporting only RocksDB. It extracts KeyedState/OperatorState operations from the serialization and deserialization logic in RocksDBStateBackend’s restore and snapshot methods.

Its drawback is that it exists outside Flink and must reimplement much of Flink’s internal savepoint handling. I tried adding an FsStateBackend reader and writer to Bravo, but the many classes and complex logic involved made me abandon the effort after finishing the reader.

It can still be used to manipulate savepoints for RocksDB-based backends.

Queryable State

Queryable State was introduced starting in Flink 1.6. Calling stateDescriptor.setQueryable(String queryableStateName) makes state queryable. It works as follows:

1. A TaskManager detecting queryable state starts a KvStateServer.
2. The sample client in the documentation queries state through ClientProxy, ServerProxy, and Server. After validation, StateTable.get(K key, N namespace) retrieves and returns the value.

With this mechanism understood, implementing a Writable State interface is straightforward: add an endpoint to KvStateServer and invoke StateTable’s write methods. The following files need changes: StateTable operations must be protected by a lock.

1
QueryableStateClient.java
2
KvStateRequest.java
3
KvStateClientProxyHandler.java
4
KvStateInternalRequest.java
5
KvStateServerHandler.java
6
StateTable.java

This approach is flexible: similar future requirements can be handled by modifying the interface. It also breaks Flink’s fault-tolerance guarantees, however. If a failure occurs after writing state but before the next checkpoint completes, Flink cannot recover that write.

The cleanest and most correct approach is to construct savepoints externally. It preserves the integrity of existing program state and introduces no unnecessary code. The Flink community is working in this direction, but substantial differences in StateBackend serialization formats and logic make universal support difficult.