Caching and Persistence

The previous three lessons gathered the source of the slowness under two headings: the data was moving and the load was unevenly distributed. The shuffle carried 1738 rows across the network and raised the total work from 2200 to 7414, the choice of join strategy put the crossover point between 1600 and 4000 rows, and the region key left two of the four workers idle and carried the skew to 3.32. In all three, what was measured was a single job running once.

On a cluster, intermediate results are not read once. One query wants the same consumption table of 2200 rows for the region total, another for the subscriber statement. This lesson asks two questions: how much it costs to do the same computation twice, and, when you decide to do that computation once and store it, whether what is stored is really there.

• PF41. The source, the first three stages of the pipeline and the distributed setup are fictional and are generated within the lesson. No distributed processing framework, cluster manager or cache product is called. The seed is 20260218.
• PF42. The cache hit, the eviction policy and the cache sizing procedure were measured in the In-Memory Stores and Caching Systems course of the Databases curriculum, and the call cache in the Building Applications with Model APIs course of the AI Engineering curriculum; they are not repeated. The only measure here is the reuse of the intermediate result.
• PF43. Makespan and total work are the model’s own unit, not raw milliseconds.
• PF44. two_queries counts the preparation as total work (all the rows) and the query step as a per-worker share. This is a mixed measure; the last section of the lesson splits it in two.
• PF45. The baseline is the setup without a cache: the intermediate result is recomputed in every query. NETWORK_COST = 3 and TASK_OVERHEAD = 50 are choices.
• PF46. Holding the intermediate result in memory is not counted as work; its cost is the held rows, and those rows are lost together with the worker that holds them.

Computing the Same Thing Twice

The first block sets up the source, the three stages of the pipeline up to consumption, and the measure of the two queries. two_queries comes from the common definition of the course; cached is its generalization and gives the same numbers at the two ends.

# FICTIONAL. Meter readings of the measurement network, the first three stages of the pipeline and
# the distributed setup are generated within the lesson; no distributed processing framework,
# cluster manager or cache product 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 = 4, 3, 50


def rows():
    v, _ = s1_screen(generate_source())
    v, _ = s2_dedupe(v)
    v, _ = s3_consumption(v)
    return v


def two_queries(records, cache=False, workers=WORKERS):
    preparation = len(records)
    if cache:
        return {"computation": preparation, "query": 2 * (preparation // workers),
                "total": preparation + 2 * (preparation // workers)}
    return {"computation": 2 * preparation, "query": 2 * (preparation // workers),
            "total": 2 * preparation + 2 * (preparation // workers)}


# ---- Generalization of the same measure: q queries , the intermediate result is alive as far as the held rows.
def cached(n, q, held, workers=WORKERS):
    """Preparation is counted as total work , the query step as a per-worker share."""
    preparation = n + (q - 1) * (n - held)
    return {"preparation": preparation, "query": q * (n // workers),
            "total": preparation + q * (n // workers)}


INTERMEDIATE = rows()
N = len(INTERMEDIATE)
print(f"intermediate result {N} rows , workers {WORKERS}")
print(f"two queries  uncached {two_queries(INTERMEDIATE)['total']} "
      f"({two_queries(INTERMEDIATE)['computation']} preparation + {two_queries(INTERMEDIATE)['query']} query)"
      f"  cached {two_queries(INTERMEDIATE, True)['total']} "
      f"({two_queries(INTERMEDIATE, True)['computation']} + {two_queries(INTERMEDIATE, True)['query']})")
print(f"generalization  {cached(N, 2, 0)['total']} and {cached(N, 2, N)['total']}")
print("\n query count     uncached      cached  diff")
for q in (1, 2, 3, 4):
    a, b = cached(N, q, 0)["total"], cached(N, q, N)["total"]
    print(f"{q:>12}  {a:>11}  {b:>10}  {a - b:>4}")

intermediate result 2200 rows , workers 4
two queries  uncached 5500 (4400 preparation + 1100 query)  cached 3300 (2200 + 1100)
generalization  5500 and 3300

 query count     uncached      cached  diff
           1         2750        2750     0
           2         5500        3300  2200
           3         8250        3850  4400
           4        11000        4400  6600

In the setup without a cache the preparation is done twice: half of the 4400 units of computation is regenerating a table that the first query has already produced. The cache deletes this second production, and the total falls from 5500 to 3300. The 2200 gained is the intermediate result’s own row count, nothing more.

The first row shows the limit of this. With a single query the setups without and with a cache pay the same 2750; the gain is zero, because there is no computation to reuse. The cache gains 6600 at the fourth query, but at the first it holds 2200 rows in memory and gives nothing in return. The gain of a cache comes not from the size of the intermediate result but from how many times it is read. Caching a result that is read once is a decision that has a cost and no gain.

When the Cache Is Not Held

The two numbers so far assumed that the cache was standing there. The second block removes that assumption. The intermediate result is distributed to four workers by the subscriber key, and each worker holds its own share in its own memory; when a worker fails, only that share is lost.

# This block continues the definitions of INTERMEDIATE, N, WORKERS, TASK_OVERHEAD and cached from the first block.
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


BUCKETS = distribute(INTERMEDIATE, "subscriber")
SHARES = [len(BUCKETS[i]) for i in range(WORKERS)]
print(f"worker shares of the intermediate result {SHARES}")
print("alive workers  held rows  total of two queries  vs uncached")
for alive in (4, 3, 2, 1, 0):
    t = sum(SHARES[:alive])
    tot = cached(N, 2, t)["total"]
    print(f"{alive:>13}  {t:>13}  {tot:>20}  {tot - 5500:>15}")
print("if a single worker fails the total "
      + " , ".join(str(cached(N, 2, N - p)["total"]) for p in SHARES))


# ---- The same setup , two separate measures: makespan (per worker) and total work.
def duration_work(n, q, held, workers=WORKERS):
    preparation = n + (q - 1) * (n - held)
    return {"duration": preparation // workers + q * (n // workers) + q * TASK_OVERHEAD,
            "total_work": preparation + q * n}


empty, full = duration_work(N, 2, 0), duration_work(N, 2, N)
print("\nmeasure         uncached      cached  gain")
for name, a, b in (("mixed total", 5500, 3300),
                   ("total work", empty["total_work"], full["total_work"]),
                   ("makespan", empty["duration"], full["duration"])):
    print(f"{name:<12}  {a:>11}  {b:>10}  {round((a - b) / a * 100, 2):>5} %")

worker shares of the intermediate result [539, 550, 561, 550]
alive workers  held rows  total of two queries  vs uncached
            4           2200                  3300            -2200
            3           1650                  3850            -1650
            2           1089                  4411            -1089
            1            539                  4961             -539
            0              0                  5500                0
if a single worker fails the total 3839 , 3850 , 3861 , 3850

measure         uncached      cached  gain
mixed total          5500        3300   40.0 %
total work           8800        6600   25.0 %
makespan             2300        1750  23.91 %

The last row of the first table is the core of the lesson: a cache that is not held at all pays exactly 5500, that is, the cost of the setup without a cache. Not a unit more and not a unit less. But the person who made the plan decided by looking at 3300; when the job is done the result that comes out is the same, the queries are correct, no check fires, and the only difference is the cost paid. This is the form that the third claim of the course takes in this lesson: what goes wrong shows up not in the output but only in the number.

The middle rows show how the gain melts away. As the held rows decrease, the total climbs linearly, and the gain is always as many as the held rows: 2200 if 2200 rows are alive, 1089 if 1089 rows are alive. A cache with three quarters alive gives three quarters of the gain. Even the failure of a single worker moves the total from 3300 to between 3839 and 3861; which worker fails makes a difference of 22 units, because the shares are not equal between 539 and 561.

Persistence is the decision beneath this table. Holding the intermediate result in memory ties it to the life of the worker; writing it to a durable place leaves the held rows at 2200 when a worker fails. In return a write job is added, and that job is paid for certain, regardless of how much of the rows will actually be read again. The gain of the cache is probabilistic, the cost of persistence is certain.

The Layout of the Cache

Up to here the cache has been treated as a set of rows. In a distributed cluster it is not so: the intermediate result sits distributed across the workers by a key, and that layout determines whether the query that will read it needs a shuffle. The third block names two queries (one asks for the region total, the other for the subscriber statement) and changes the layout in which the cache is held.

# This block continues the definitions of INTERMEDIATE, N, BUCKETS, WORKERS, NETWORK_COST, trace_id
# and distribute from the first two blocks.
def shuffle(buckets, new_key, workers=WORKERS):
    new = {i: [] for i in range(workers)}
    moved = 0
    for old, records in buckets.items():
        for k in records:
            target = trace_id(k[new_key]) % workers
            if target != old:
                moved += 1
            new[target].append(k)
    return new, moved


BY_REGION, sub_to_region = shuffle(BUCKETS, "region")
_, region_to_sub = shuffle(BY_REGION, "subscriber")
print(f"moved from the subscriber layout to region {sub_to_region} , from the region layout to subscriber {region_to_sub}")
print("cache layout        held rows  shuffled in 4 queries  network work")
for name, held, moved in (("subscriber", N, 2 * sub_to_region), ("region", N, 2 * region_to_sub),
                          ("two copies", 2 * N, sub_to_region)):
    print(f"{name:<18}  {held:>9}  {moved:>21}  {moved * NETWORK_COST:>12}")

moved from the subscriber layout to region 1738 , from the region layout to subscriber 1738
cache layout        held rows  shuffled in 4 queries  network work
subscriber               2200                   3476         10428
region                   2200                   3476         10428
two copies               4400                   1738          5214

Both directions move the same 1738 rows; in which layout it is stored does nothing other than choose which query will be free. If two of the four queries ask for the region and two for the subscriber, the cache with a single layout moves 3476 rows and, at NETWORK_COST = 3, produces 10428 units of network work. Holding two copies brings this number down to 5214, because the second copy is built once and is never shuffled afterward.

The cost is visible in memory: the held rows rise from 2200 to 4400, and according to the table of the previous section, every worker that fails now takes a share of both copies with it. The cache here is not a set of rows but a set of rows plus a layout; storing a second layout is holding a second cache.

Which Number Fell

The last table of the second block writes a single cache decision with three measures, and all three are correct. The mixed total shows a gain of 40%, because it counts the preparation with all the rows and the query with a per-worker share. Total work shows 25%: in the figure that falls from 8800 to 6600, the query step is also counted with all the rows, and the cache does not touch it. Makespan shows 23.91%; the preparation per worker falls from 1100 to 550, but every query still pays its own TASK_OVERHEAD = 50.

The same change has three separate speedup numbers, and the largest gap among them is 16%. For a setup, the sentence “the cache sped the job up by forty percent” is not a verifiable claim unless it is written which number fell. Makespan and total work are measured separately; in this lesson the two went in the same direction, but that is not a rule, it is the result of this setup.

Summary

• The two queries that share an intermediate result of 2200 rows total 5500 without a cache and 3300 with one; the 2200 gained is the intermediate result’s own row count.
• With a single query both setups pay 2750. The gain of a cache comes not from the size of the intermediate result but from how many times it is read; caching a result that is read once is a decision that has a cost and no gain.
• The intermediate result sits in four workers as 539, 550, 561 and 550 rows. As workers fail, the total rises from 3300 to 3850, 4411, 4961 and 5500; a cache that is not held at all pays exactly the cost of the setup without a cache.
• The gain is always as many as the held rows, and the failure of a single worker moves the total to between 3839 and 3861. A durable write leaves the held rows at 2200, and in return the write cost is paid for certain.
• A cache is not a set of rows but a set of rows plus a layout: both directions move the same 1738 rows, with four queries a single layout produces 3476 rows and 10428 units of network work, and two copies bring this down to 5214 while raising the held rows from 2200 to 4400.
• The same cache shows a gain of 40% in the mixed total, 25% in total work and 23.91% in makespan; a claim of speedup that does not say which was measured is incomplete.

Next Step

The last table of this lesson counted the failure of a worker only as the loss of the cache: what disappeared was a stored intermediate result, and it could be recomputed. In a job that flows continuously, however, what sits in the worker’s memory is not an intermediate result but the state of the computation up to that moment, and when it is to be recomputed, it is not clear where to start. The next lesson counts how much a failed worker asks to have back: how many rows are recomputed in a job that never saves the state, how far raising the checkpoint frequency lowers this number, and what it costs to prevent the same row from being counted twice.