State in Stream Engines

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.

# 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.

# 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.

# 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.