Stateful processing¶

A TaskManager disk-usage alert fires: the RocksDB directory has hit 380 GB and is still climbing, on a job that does nothing more exotic than count failed logins per user_id.

A. Key skew — one whale customer_id is holding most of the state. B. No TTL — state for users who logged in successfully months ago is still sitting there. C. The job stores a ListState of every raw event instead of a small counter. D. A broadcast dimension table is too large and is duplicated on every TaskManager.

Predict which one before reading on — then work out how many bytes per key you would expect if the diagnosis is right.

Start with the situation¶

Windows are state with an opinionated API. Plenty of jobs need memory of the past without a window:

• "How many failed logins has this user_id had since the last success?"
• "What was the previous latency_ms for this endpoint?" (delta)
• "Is this IoT device_id in the same session?"
• E-commerce: join a click stream with a broadcast table of fraud rules that update every minute.

A stateless map/filter cannot do this. A stateful operator stores bytes keyed by something, checkpoints them (checkpoints), and restores after a crash.

Why the obvious approach breaks at scale¶

State is a database you accidentally wrote inside the pipeline.

10 million users × 100 bytes looks like 1 GB. Then you add a MapState of 30 days of days. Then you forget TTL. Then RocksDB holds 400 GB on each TaskManager's local SSD, checkpoints take 25 minutes, and a rescale means 400 GB reshuffle.

Keyed state is sharded by key hash onto parallel subtasks — the same skew story as Kafka partitions. One customer_id whale holds a giant ListState on one slot.

Build the mental picture¶

After key_by(user_id), each operator subtask owns a slice of keys. For each key you may keep:

• ValueState — one value (the count)
• ListState — append-only list (careful)
• MapState — map inside the key (e.g. endpoint → count)
• ReducingState / AggregatingState — incremental

Timers (register_event_time_timer) fire when the watermark (or processing time) reaches a timestamp — how you implement "reset after 5 minutes" without a window.

Operator state is not keyed: each subtask has its own list (Kafka offsets in older sources, buffers). Broadcast state is a replicated map every parallel instance sees (rules, dimension tables).

Internals: keyed vs operator vs broadcast¶

Keyed state¶

from pyflink.datastream import KeyedProcessFunction
from pyflink.datastream.state import ValueStateDescriptor
from pyflink.common import Types

class FailedLoginCounter(KeyedProcessFunction):
    def open(self, runtime_context):
        self.count = runtime_context.get_state(
            ValueStateDescriptor("failed_login_count", Types.INT())
        )

    def process_element(self, event, ctx: KeyedProcessFunction.Context):
        n = self.count.value() or 0
        if event["endpoint"] == "/login" and event["status_code"] in (401, 403):
            n += 1
            self.count.update(n)
            if n > 10:
                yield f"ALERT {event['user_id']} count={n}"
        elif event["status_code"] == 200 and event["endpoint"] == "/login":
            self.count.clear()

State for u_99102 never collides with u_12. Parallelism can increase if you restore from a savepoint (Flink redistributes keys). Changing the key (user_id → customer_id) makes old state unreachable.

Operator state¶

Per-subtask, not per-key. Used by sources (split assignment), custom connectors, random buffers. On restore, Flink redistributes operator list state with a routing scheme (even-split, union). Getting this wrong duplicates Kafka splits.

Broadcast state¶

A control stream (fraud rules, "this customer_id is a whale") is broadcast to all parallel instances of a BroadcastProcessFunction. The keyed event stream reads the broadcast map. Rules must fit in memory on every TaskManager.

Internals: state backends¶

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.state_backend import EmbeddedRocksDBStateBackend, HashMapStateBackend

env = StreamExecutionEnvironment.get_execution_environment()
env.set_state_backend(EmbeddedRocksDBStateBackend(True))  # incremental checkpoints

RocksDB: each state access may be a disk read (with memory cache). Expect 10–100× slower than heap for random ValueState access. Incremental checkpoints only work with RocksDB (and a few others); they upload SST diffs rather than the full state each time.

Heap backend: one GC pause is a latency incident; a 20 GB heap of state will not go well.

Internals: TTL and timers¶

Unbounded keyed state is a leak.

from pyflink.datastream.state import StateTtlConfig, ValueStateDescriptor
from pyflink.common import Duration, Types

ttl = (
    StateTtlConfig.new_builder(Duration.of_hours(24))
    .set_update_type(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .set_state_visibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanup_full_snapshot()
    .build()
)
desc = ValueStateDescriptor("count", Types.INT())
desc.enable_time_to_live(ttl)

TTL cleanup is not instant; RocksDB compaction / snapshot cleanup eventually drops it. Timers you register are also state. A timer per event (instead of per key) is a well-known OOM.

Event-time timers fire on watermark progress — idle Kafka partitions that stall watermarks also stall timers (time).

How: session-ish fraud without a window¶

class RollingFailures(KeyedProcessFunction):
    """Count failures; expire the count 5 minutes after the last failure (event time)."""

    def open(self, ctx):
        self.count = ctx.get_state(ValueStateDescriptor("n", Types.INT()))
        self.last_ts = ctx.get_state(ValueStateDescriptor("t", Types.LONG()))

    def process_element(self, event, ctx: KeyedProcessFunction.Context):
        ts = event["ts_ms"]
        n = self.count.value() or 0
        if event["status_code"] in (401, 403):
            n += 1
            self.count.update(n)
            self.last_ts.update(ts)
            ctx.timer_service().register_event_time_timer(ts + 5 * 60 * 1000)
            if n > 10:
                yield event
        else:
            self.count.clear()

    def on_timer(self, timestamp, ctx: KeyedProcessFunction.OnTimerContext):
        last = self.last_ts.value()
        if last is not None and timestamp >= last + 5 * 60 * 1000:
            self.count.clear()
            self.last_ts.clear()

This is exact "last 5 minutes" only if you also decrement on timer for each failure (a list of timestamps). The sketch above is a reset-after-quiet approximation — know which one product wanted.

Where teams get caught¶

• ListState of raw events for a 1-hour window: you stored the stream twice. Use AggregateFunction.
• Changing state descriptor names without a migration: restore finds empty state. Use @type_info / StateMigration / savepoint with schema evolution (POJOs/Avro, not untyped Python tuples if you care).
• Python / PyFlink state crosses to the JVM; hot paths at 2M/s are Java/Scala operators.
• TTL OnReadAndWrite vs OnCreateAndWrite: a noisy read-only key lives forever with the former.
• Broadcast a 5 GB dimension table: every TM holds 5 GB. Use a regular keyed join or an external store.

How it fails¶

RocksDB vs heap: a concrete latency picture¶

Heap ValueState is a hash lookup — tens of nanoseconds if the object is hot. RocksDB is a get that may hit block cache (microseconds) or SSD (tens to hundreds of microseconds). A KeyedProcessFunction that does five state reads per event at 100k events/s/core is trivial on heap and a CPU+IO budget on RocksDB.

That is why fraud counts (tiny value, huge cardinality) still work on RocksDB, while "store the last 100 events as ListState" does not. Access pattern × cardinality × backend is the capacity model.

Measure ValueState access time in the Flink metrics (RocksDB actual-user-key histograms if enabled) before you "optimise" the algorithm. Many jobs are sink-bound; RocksDB is innocent.

How to investigate¶

If checkpoint size of FailedLoginCounter is 80% of the job, dump key cardinality: unique user_id per day vs TTL.

Scale: 10× / 100× / 1000׶

Assume 8 bytes count + 32 bytes key overhead ≈ 50 bytes/key (the exercise in the original notes).

7-day TTL at 10M users/day ≈ 70M keys × 50 B ≈ 3.5 GB logical plus RocksDB files. Checkpoint duration becomes the SLO.

Trade-offs¶

Alternatives¶

• Kafka Streams state stores (also RocksDB, changelog topics) — same idea, embedded.
• Redis / DynamoDB as the store, Flink stateless — operationally simpler at modest QPS; you lose local checkpoint atomicity unless you design idempotent writes.
• Spark mapGroupsWithState / flatMapGroupsWithState — micro-batch state; see comparison.

Joins are state¶

A stream-stream interval join buffers records from both sides until the watermark says they cannot match. That buffer is keyed state. A 5-minute join window at 200k clicks/s is a large RocksDB.

A stream-table join against a compacted user-profile topic keeps the latest profile per user_id as ValueState (or a KTable in Kafka Streams). Cardinality = users, not events — usually the right shape.

Broadcast join: rules file 2 MB, every TM holds it. Fine. Broadcast a 20 GB item catalogue: you have designed a distributed out-of-memory error.

Querying Postgres from process_element is not Flink state. It is an RPC. It will not be snapshotted. It will amplify QPS by the event rate. Cache with TTL in state if you must, and accept staleness.

Rescaling¶

Heap and RocksDB keyed state can be redistributed on restore from a savepoint (or from a retained checkpoint when the deployment supports it). The unit is a key group, not an individual key. Runtime parallelism may be any positive value up to maxParallelism; it does not need to divide it evenly. Set maxParallelism deliberately before the first state you must preserve, because changing it changes key-to-group assignment and normally prevents compatible restore.

Operator state (source splits) rescales with a different contract. Never assume a custom source survived a 4→12 rescale without a test.

Changelog thinking (why Kafka Streams feels similar)¶

Every keyed state update can be emitted as a changelog: (key, new_value) or a retraction. Kafka Streams does this onto compact topics so a restarted instance can rebuild RocksDB. Flink does it onto checkpoint storage instead (unless you use queryable state / SQL upsert to Kafka).

The implication for this academy: if you need another job to read "current failed-login count per user", sink a compacted changelog topic from Flink rather than poking TaskManager memory. That topic is the compaction use case — latest value per user_id.

Do not log every state update to a JSON debug topic at 2M/s. You will DDoS Kafka and learn nothing.

Serialisation of state¶

Heap backend keeps Java objects (or PyFlink's representation). RocksDB needs bytes. A change to the Python tuple layout without a TypeSerializer migration yields silent corruption or restore failure.

Prefer explicit Types.INT(), Types.STRING(), POJOs, Avro for anything you will restore in six months. pickle in state is a future incident.

Queryable state (Flink) lets other services hit a TM over RPC for the current count. It is convenient and operationally fragile (you are now serving production reads from the processor). The robust pattern is still: compacted Kafka changelog or a serving DB populated by the sink.

For IoT, shard state by device_id prefix onto separate jobs if a single RocksDB instance cannot compact fast enough. That is a 1000× move, not a 10× one. At 10×, TTL and a single job are enough.

How: RocksDB job config worth copying¶

config = Configuration()
config.set_string("state.backend.incremental", "true")
config.set_string("state.backend.rocksdb.memory.managed", "true")
config.set_string("state.checkpoints.dir", "s3://data-platform/flink/checkpoints")
env = StreamExecutionEnvironment.get_execution_environment(config)
env.set_state_backend(EmbeddedRocksDBStateBackend(True))

Local RocksDB directories belong on NVMe, not on a shared network volume. NFS-backed rocksdb is a classic "why is everything 20× slower" ticket.

How to apply this at work¶

Inventory every get_state / ValueStateDescriptor in the job. For each: key, estimated cardinality, bytes per key, TTL, backend. If nobody can estimate cardinality, you do not have a state budget.

Check your understanding¶

10 million unique users/day. Keyed ValueState of one float per user (~50 bytes/pair including overhead). RocksDB.

1. State after 1 day, no TTL?
2. Steady-state with 7-day TTL?
3. When do you worry about checkpoint duration?
4. What RocksDB knobs matter first?