--- title: 'State in Stream Engines' source: 'https://academia.sh/en/courses/distributed-data-processing/state-in-stream-engines' course: 'Distributed Data Processing' language: en updated: '2026-10-21T05:03:00+00:00' license: 'CC BY-SA 4.0' objectives: - 'Counts the rows recomputed after a failure in a streaming job by checkpoint interval.' - 'Compares the certain cost of a checkpoint with the probabilistic gain of recovery and justifies the interval decision by the failure probability.' - 'Calculates the rows a single failed worker asks to have back under partitioned state and measures the state that exactly-once semantics holds.' --- # State in Stream Engines Counting how much a failed worker asks to have back in a job that flows continuously, and whether raising the checkpoint frequency really lowers that number: in a streaming job of 2200 rows, when a failure occurs in period 2026-09, 1600 rows are recomputed without a checkpoint, 400 at interval 6, again 400 at interval 3, and 0 at interval 1. Halving the interval does not halve the recovery, because the last point before the failure has not moved; interval 4 gives 0 on the same failure. When all eleven failure points are scanned, interval 6 makes 545.5 rows on average be recomputed and interval 3 makes 218.2, but the certain cost rises from 400 to 800; by the worst case interval 4 and 3 tie, by the average interval 4 wins. When the failure probability drops from 1 to 0.5, the best choice turns back to holding no checkpoint. With partitioned state a single failed worker asks for 408 rows instead of 1600 and 102 instead of 400, and exactly-once requires holding exactly that many rows of state. The previous lesson counted a failed worker only as **the loss of the cache**: what disappeared was a stored intermediate result, it could be recomputed, and the total of two queries rose from 3300 to 5500. That is what is right in a batch job, because the input is still standing where it was. In a job that flows continuously the situation is different. What sits in the worker's memory is not an intermediate result but **the state** of the computation up to that moment: which subscriber has been reached up to which period, where which partial total has stopped. When a failure occurs, this state is lost and the job has to start again **from somewhere**. This lesson counts a single question: how much the failed worker asks to have back, and whether the cost paid to shrink that number really pays for itself. - **PF47.** The source, the pipeline and the distributed setup are **fictional** and are generated within the lesson. No stream engine, cluster manager or state store is called. The seed is 20260218. - **PF48.** The procedures for the checkpoint, resilience to duplicates and exactly-once semantics were measured in the Data Ingestion topic of the Data Pipelines and Orchestration course of the Data Engineering and MLOps curriculum and in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum; they are **not repeated**. The numbers there are **input** here: 2200 rows at least once with 19 duplicates, 2162 rows at most once. The only measure here is **the cost of recovery**. - **PF49.** The streaming job advances period by period and **the failure occurs in period `2026-09`**; the job continues from the last checkpoint before the failure. In the code the `checkpoints` field carries the number of checkpoints. - **PF50.** A checkpoint makes the current state durable; the state is **one entry per subscriber**, that is, `STATE_ROWS = 200` rows. This is a choice, and all the totals below are sensitive to it. - **PF51.** When averaging, the failure is assumed to occur **with equal probability in any one** of the eleven periods; this too is a choice. The row is the model's own unit, not raw milliseconds. - **PF52.** The baseline is the setup **that holds no checkpoint**: point 0, certain cost 0, and after a failure it starts over from the beginning. ## The Cost of Recovery The first block sets up the source, the three stages of the pipeline up to consumption, and the `recovery` measure. `recovery` counts how many of the rows up to the failure period will be recomputed. ```python # FICTIONAL. Meter readings of the measurement network, the first three stages of the pipeline and # the streaming job are generated within the lesson; no stream engine , cluster manager # or state store is called. Seed 20260218. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue out.append(dict(k, m3=int(k["index"]) - int(prev["index"]))) return out, dropped WORKERS, NETWORK_COST, TASK_OVERHEAD, STATE_ROWS = 4, 3, 50, 200 def rows(): v, _ = s1_screen(generate_source()) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v def recovery(records, checkpoint_interval, failed_period="2026-09", workers=WORKERS): """Without a checkpoint , from the start; with one , from the last point.""" i = PERIODS.index(failed_period) if checkpoint_interval is None: recomputed = sum(1 for k in records if PERIODS.index(k["period"]) <= i) return {"checkpoints": 0, "recomputed": recomputed} last = (i // checkpoint_interval) * checkpoint_interval recomputed = sum(1 for k in records if last < PERIODS.index(k["period"]) <= i) return {"checkpoints": len(PERIODS) // checkpoint_interval, "recomputed": recomputed} STREAM = rows() print(f"stream job {len(STREAM)} rows , failure 2026-09") print("interval checkpoints recomputed rows") for a in (None, 6, 3, 1): s = recovery(STREAM, a) print(f"{str(a):>8} {s['checkpoints']:>11} {s['recomputed']:>15}") ``` ``` stream job 2200 rows , failure 2026-09 interval checkpoints recomputed rows None 0 1600 6 2 400 3 4 400 1 12 0 ``` Three of the four rows are what one would expect. The job that holds no checkpoint recomputes **all 1600 rows** up to the failure period; interval 6 lowers this to 400; interval 1 lowers it to zero because it saves every period. The third row is not what one would expect: **going from interval 6 to interval 3 does not change the recomputed rows at all.** The interval has been cut in two, the number of held points has gone from 2 to 4, and the recovery has stayed at 400. What is paid has doubled; what is received has stayed the same. ## Halving the Interval The reason stands in a single calculation. The failure is `2026-09`, that is, the eighth in the index of twelve. At interval 6 the last point before the failure is the sixth period; at interval 3 it is also the sixth period, because the remainder of dividing eight by three, two, carries the last point back to six. **The interval was halved but the last point before the failure did not move.** The second block scans this behavior across all eleven failure points. ```python # This block continues the definitions of STREAM, PERIODS, STATE_ROWS and recovery from the first block. INTERVALS = (None, 6, 4, 3, 2, 1) print("failure " + " ".join(f"{str(a):>5}" for a in INTERVALS)) for d in PERIODS[1:]: print(f"{d} " + " ".join(f"{recovery(STREAM, a, d)['recomputed']:>5}" for a in INTERVALS)) print("\ninterval checkpoints certain cost worst worst total average average total") for a in INTERVALS: y = [recovery(STREAM, a, d)["recomputed"] for d in PERIODS[1:]] k = recovery(STREAM, a)["checkpoints"] * STATE_ROWS o = sum(y) / len(y) print(f"{str(a):>8} {recovery(STREAM, a)['checkpoints']:>11} {k:>12} {max(y):>5}" f" {max(y) + k:>11} {round(o, 1):>7} {round(o + k, 1):>13}") ``` ``` failure None 6 4 3 2 1 2026-02 200 200 200 200 200 0 2026-03 400 400 400 400 0 0 2026-04 600 600 600 0 200 0 2026-05 800 800 0 200 0 0 2026-06 1000 1000 200 400 200 0 2026-07 1200 0 400 0 0 0 2026-08 1400 200 600 200 200 0 2026-09 1600 400 0 400 0 0 2026-10 1800 600 200 0 200 0 2026-11 2000 800 400 200 0 0 2026-12 2200 1000 600 400 200 0 interval checkpoints certain cost worst worst total average average total None 0 0 2200 2200 1200.0 1200.0 6 2 400 1000 1400 545.5 945.5 4 3 600 600 1200 327.3 927.3 3 4 800 400 1200 218.2 1018.2 2 6 1200 200 1400 109.1 1309.1 1 12 2400 0 2400 0.0 2400.0 ``` The matrix shows by name where the gain comes from: **from the distance of the failure to the last point.** In `2026-07` interval 6 gives zero, because the failure has landed exactly on the point; the same interval gives 1000 in `2026-12`. Interval 4 gives 0 in `2026-09` while interval 3 gives 400 there; the setup that holds points more often is worse in that failure. None of the six columns beats the others in every row. The average column erases this noise and behaves regularly: 1200.0 → 545.5 → 327.3 → 218.2 → 109.1 → 0. **On average, halving the interval roughly halves the recovery; in a single failure it does not.** A setup experiences a single failure, not the average; whoever decides by the average meets the failure that actually happens. The last two columns add the cost. Holding a checkpoint pays a certain cost over `STATE_ROWS = 200`: 400 at interval 6, 800 at interval 3, 2400 at interval 1. By the worst case the cheapest setup is **interval 4 and interval 3 with 1200**, and by the average **interval 4 with 927.3**. The two criteria give two different answers, and the rule "halve the interval" is the best in neither. ## Certain Cost and Probabilistic Gain The totals above carry a silent assumption: that there is a failure in every run. The third block relaxes this assumption and gives the same decision again at four failure probabilities. Then it splits the job across four workers and counts how much the single failed worker asks to have back. ```python # This block continues the definitions of STREAM, PERIODS, WORKERS, STATE_ROWS, INTERVALS and # recovery from the first two blocks. def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h def distribute(records, key, workers=WORKERS): buckets = {i: [] for i in range(workers)} for k in records: buckets[trace_id(k[key]) % workers].append(k) return buckets print("failure probability " + " ".join(f"{str(a):>6}" for a in INTERVALS) + " best") for p in (1.0, 0.75, 0.5, 0.25): s = [] for a in INTERVALS: y = [recovery(STREAM, a, d)["recomputed"] for d in PERIODS[1:]] s.append(p * sum(y) / len(y) + recovery(STREAM, a)["checkpoints"] * STATE_ROWS) print(f"{p:>19} " + " ".join(f"{round(v, 1):>6}" for v in s) + f" {str(INTERVALS[s.index(min(s))]):>6}") BUCKETS = distribute(STREAM, "subscriber") print("\ninterval whole job from scratch busiest worker with partitioned state exactly-once state") for a in (None, 6, 3, 1): g = recovery(STREAM, a)["recomputed"] per = max(recovery(BUCKETS[i], a)["recomputed"] for i in range(WORKERS)) print(f"{str(a):>8} {g:>22} {per:>37} {g:>18}") ``` ``` failure probability None 6 4 3 2 1 best 1.0 1200.0 945.5 927.3 1018.2 1309.1 2400.0 4 0.75 900.0 809.1 845.5 963.6 1281.8 2400.0 6 0.5 600.0 672.7 763.6 909.1 1254.5 2400.0 None 0.25 300.0 536.4 681.8 854.5 1227.3 2400.0 None interval whole job from scratch busiest worker with partitioned state exactly-once state None 1600 408 1600 6 400 102 400 3 400 102 400 1 0 0 0 ``` The first table shows that the interval decision depends **not on the workload but on the failure frequency**. If there is a failure in every run the best choice is interval 4; at a probability of three quarters it moves to interval 6; when the probability is halved the best choice is **to hold no checkpoint at all**. The reason is one sentence: **the cost of a checkpoint is certain, its gain is probabilistic.** The certain cost is paid in every run; the gain materializes only when a failure occurs and only as much as the distance of the failure to the last point. ## How Much the Failed Worker Asks to Have Back The second table answers the opening question of the lesson. If the job is split across four workers by the `subscriber` key and the state is held at each worker **for its own partition**, the single failed worker asks to have back not the whole job but only its own share: **408** rows instead of 1600, **102** instead of 400. Partitioned state lowers the recovery to a quarter, but only on the condition that the state is held partitioned. If the state is gathered in one place, or if the engine restarts the whole job on a failure, the number asked back is again 1600 and 400. **Partitioning makes recovery cheaper only if the state is partitioned too.** The last column is the cost of **exactly-once** semantics, and its number is surprisingly familiar. The rows recomputed during recovery are written to the output **a second time**; the only way to prevent this is to remember which rows have been written since the last point. That set is exactly the set of recomputed rows: 1600 when there is no interval, 400 at intervals 6 and 3, 0 at interval 1. **Exactly-once is holding as much state as the rows the recovery recomputes**; making the interval denser makes recovery cheaper while it also shrinks this state, and the two gains come from the same number. This set is itself a state, and it is lost together with the failed worker. If the list of written rows stands only in memory, the recovery loses that too, and which row went to the output becomes unknown; exactly-once is achieved only if the list is taken into the checkpoint. This carries the certain cost beyond `STATE_ROWS = 200`: the point now holds not only the partial totals but also the identities of the rows written since the last point. At interval 6 that is 400 identities, and at interval 3 also 400. What happens if this is not held was measured in the Data Pipelines and Orchestration course: the at-least-once setup lets 2200 rows through, 19 duplicates occur and the reported deviation **stays empty**. The third claim of the course holds here too: as long as exactly-once is achieved, all four of the four intervals produce **the same result**. There is no way to tell from the result which interval was chosen; the difference is only in the cost paid. ## Summary - In the streaming job of 2200 rows, on a failure in `2026-09` the recomputed rows are **1600** with no checkpoint, **400** at interval 6, **again 400** at interval 3, and **0** at interval 1. - Halving the interval does not halve the recovery: the last point before the failure has not moved. On the same failure interval 4 gives **0** and interval 3 gives **400**; the setup that holds points more often can be worse. - The average over the eleven failure points is regular (1200.0 → 545.5 → 327.3 → 218.2 → 109.1 → 0), but a setup experiences a single failure, not the average. By the worst case the cheapest is interval 4 and 3 with 1200, by the average interval 4 with 927.3. - The cost of a checkpoint is certain, its gain is probabilistic: at failure probability 1 the best interval is 4, at 0.75 it is 6, and at 0.5 and below it is **to hold no checkpoint**. - With partitioned state the single failed worker asks for **408** rows instead of 1600 and **102** instead of 400; partitioning makes recovery cheaper only if the state is partitioned too. - Exactly-once is holding as much state as the rows the recovery recomputes: 1600, 400, 400 and 0. As long as this is achieved, all four intervals produce the same result and the difference is only in the cost. ## Next Step Five lessons wrote five separate names of the cost: the shuffled row, the copy, the skew, the held row and the recomputed row. Beneath all of them stood the same two numbers, **how long a job took** and **how much work it did in total**, and in every lesson these two were written separately. One question remains: with how many workers a cluster should be set up. The next lesson sweeps the number of workers from one to sixteen and shows that the total work rises while the makespan falls; it counts that this decision cannot be made without writing which one is paid, and closes the course.