In Flink, why prefer MapState over a ValueState holding a HashMap?

## The two shapes of "a map per key" You need to keep, for each keyed entity, a set of sub-keyed facts: per-user, a map of product id to last-seen timestamp. Flink lets you express that two ways. You can declare `MapState<String, Long>` with a `MapStateDescriptor`, or you can declare `ValueState<Map<String, Long>>` with a `ValueStateDescriptor` and put a `HashMap` inside it. They look interchangeable in the code. They are not interchangeable at runtime, and the reason is how each state backend physically stores a value. ## Why the backend decides the answer Flink bundles `HashMapStateBackend` and `EmbeddedRocksDBStateBackend`. `HashMapStateBackend` holds state internally as ordinary Java objects on the TaskManager heap. Reading a `ValueState` hands you back the very object you stored — no serialisation on the access path at all. So a `HashMap` inside a `ValueState` costs roughly the same as `MapState` for point access; the difference is mostly about API convenience and snapshot size. (Because objects are shared rather than copied, this backend is also the one where mutating a value you read out is unsafe.) `EmbeddedRocksDBStateBackend` is different in kind. It keeps in-flight state in an embedded RocksDB instance in the TaskManager's local data directories, and everything is stored as **serialised byte arrays**, with key comparisons done byte-wise rather than through `hashCode()`/`equals()`. Every read has to deserialise; every write has to serialise. That is the whole reason RocksDB scales past heap, and it is also the reason the granularity of a state entry matters. ## The cost model, concretely Under RocksDB with `ValueState<Map<String, Long>>` containing 5,000 entries: - `map = state.value()` deserialises 5,000 pairs to build the `HashMap`. - You mutate one entry. - `state.update(map)` re-serialises all 5,000 pairs and writes one large value. Do that once per record and per-record work is proportional to the size of the map, not to the size of the change. Throughput collapses as the map grows, and RocksDB is doing large-value writes that inflate its write amplification. With `MapState<String, Long>`, Flink composes the backend's storage key from the keyed-state key *and* the user map key, so each pair is its own RocksDB entry: - `state.get("p-99")` deserialises one value. - `state.put("p-99", ts)` writes one small entry. Per-record work is now independent of map size. `remove(uk)` deletes one entry, `contains(uk)` probes one entry, and `isEmpty()` avoids pulling anything into the JVM. Only the iterating methods — `entries()`, `keys()`, `values()` — walk the whole map, and they stream lazily rather than materialising a copy. ## Two more differences that matter in production **TTL granularity.** All of Flink's state collection types support per-entry TTLs: list elements and map entries expire independently. So with `MapState` plus a `StateTtlConfig`, a stale product entry disappears on its own while the rest of the user's map survives. With `ValueState<Map<...>>` the blob is one state entry: either the whole map is still live or the whole map expires. If your reason for keeping the map is "remember recent things and forget old ones", `MapState` is doing the forgetting for you and `ValueState` is not. **Size limits.** RocksDB's JNI bridge is `byte[]`-based, so the maximum supported size is 2^31 bytes per key and per value. A `ValueState` holding a very large map is a single value marching towards that ceiling. Related: states implemented with RocksDB merge operations, such as `ListState`, can silently accumulate past 2^31 bytes and then fail on the next retrieval — another reason unbounded per-key collections are a design smell. ## Where ValueState-of-map is still fine It is not always wrong. If the map is small and bounded — a handful of counters, a fixed set of flags — and you almost always read or write all of it together, the blob form is simpler and can even be cheaper: one serialised value instead of several backend entries, and one write instead of several. If you are on `HashMapStateBackend` and your state comfortably fits in heap, the performance argument largely evaporates, though the TTL-granularity argument does not. The rule of thumb: if the collection's size grows with input, or you access one entry at a time, use `MapState`. If it is a fixed-size record you touch as a unit, a `ValueState` of a small object is fine — and often a POJO is a better choice than a `Map`, because POJO types are one of the two families whose schema Flink can evolve across a savepoint. ## Migrating an existing job Switching a running job from `ValueState<Map<K,V>>` to `MapState<K,V>` is *not* a schema evolution Flink performs for you — the state name and structure change, so the old bytes are simply not read. Either give the new state a different name and accept starting empty, or take a savepoint and rewrite it offline with the State Processor API.