--- title: 'Caching and Persistence' source: 'https://academia.sh/en/courses/distributed-data-processing/caching-and-persistence' course: 'Distributed Data Processing' language: en updated: '2026-10-21T05:03:00+00:00' license: 'CC BY-SA 4.0' objectives: - 'Counts the total cost of queries that share an intermediate result, with and without a cache, and determines the number of queries at which the cache gives no gain.' - 'Measures the loss of an intermediate result held in memory when a worker fails and shows how far an unheld cache misleads the plan.' - 'Writes the same cache gain separately in the measures of mixed total, total work and makespan.' --- # Caching and Persistence Counting the cost of computing the same intermediate result twice when more than one query reads it, and the cost that comes back when the cache is not held: two queries sharing an intermediate result of 2200 rows total 5500 without a cache and 3300 with one, but with a single query both are 2750 and the cache gains nothing. The intermediate result sits in the memory of four workers as 539, 550, 561 and 550 rows; as workers fail, the total rises from 3300 to 3850, 4411, 4961 and 5500, and a cache that is not held at all pays exactly the cost of the setup without a cache, although the plan was made for 3300. The failure of a single worker moves the total to between 3839 and 3861. The same cache shows a gain of 40% in the mixed total, 25% in total work and 23.91% in makespan, and a claim of speedup is incomplete unless it is written which of the three numbers was measured. 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. ```python # 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. ```python # 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. ```python # 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.