--- title: 'Resilience to Repetition and Reprocessing' source: 'https://academia.sh/en/courses/introduction-to-data-engineering/resilience-to-repetition-and-reprocessing' course: 'Introduction to Data Engineering' language: en updated: '2026-10-21T05:03:01+00:00' license: 'CC BY-SA 4.0' objectives: - 'Runs idempotent and non-idempotent write modes on the same reprocessing scenario and counts the duplicated rows.' - 'Shows that the deduplication stage absorbs the surplus and that the served value therefore does not change while the raw row count triples over three runs.' - 'Measures that when the deduplication key is chosen wrongly, the same rerun grows the served value together with the number of runs.' --- # Resilience to Repetition and Reprocessing Measuring that processing the same load a second time duplicates rows and that the served number does not show it: in an appending store, backfilling one period takes the raw rows from 2400 to 2600 and a full rerun takes them to 4800, reaching 7200 in three runs, yet the served value stays at 1212.55 in all three runs because the deduplication stage absorbs the surplus; when the resent load carries a correction, the same stage throws the correction away and leaves the value at 794.36; and when deduplication looks at the whole row instead of the key, the same reruns carry the value to 2425.09 and 3637.64. The previous lesson showed that the contract catches the corruption at stage 0, but catching is not repairing: the 200 blocked rows wait somewhere, the producer fixes them, and that period is **resent** to the pipeline. In none of the measurements so far has the pipeline seen the same data twice. The second run is not an exception; it is the most frequent thing a pipeline does. A producer corrects a load, a stage fails and is retried, a past period is **backfilled**. This lesson counts what happens to the row count of a pipeline that processes the same load twice, which stage absorbs the surplus, and whether the served value shows it. The answer is the cleanest example of the third claim of the course. - **QP14.** The source, the pipeline and the store are **fictional**; no real store, file format or orchestrator is called. The seed is **20260218**, and the measurement does not depend on the run. - **QP15.** The equivalent of `idempotent` in this curriculum is **idempotent**. The concept itself was measured in the Shell Programming course, and the retry and queue procedure in the Caching, Queues and Asynchronous Processing course, and **is not repeated**. The only question here is **whether reprocessing duplicates the row**. - **QP16.** Reprocessing is not an accident: a corrected load, a retried stage and a backfilled period are ordinary jobs. - **QP17.** The **value of a resent row stays the same**, only the load stamp changes; the one exception is the load that carries a correction. This detail determines the whole of the last two measurements. - **QP18.** The baseline is the **single run**: raw rows 2400, duplicated rows **0**. - **QP19.** The units counted: reprocessed rows, duplicated rows, rows dropped by deduplication and the difference at serving. **No raw time is written**, the unit is written as **run**. - **QP20.** Resolution: the step of the row share is **0.0004**, of a summary cell **0.0182**, of the serving value **0.2000**. **No share is printed on the serving output.** ## Two Write Modes Whether a pipeline is idempotent is decided by a single choice: **how** the incoming row is written to the store. An **appending** write puts the row at the end of the store; when the same row arrives twice, it sits in the store twice. A **keyed** write overwrites the row at a key; when the same key arrives twice, it sits in the store once. The key here is the pair of subscriber and period, which is already what the pipeline uses as its uniqueness criterion. The second part is how reprocessing looks. When a corrected load arrives, the **value of the row is the same**; the only thing that changes is the load stamp. This detail is often overlooked, and the whole of the last measurement rests on it. ```python # FICTIONAL. Meter readings, the five-stage pipeline and the store are defined in the lesson; # no real store, file format or orchestrator is called. 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 = ["east", "west", "north", "south", "central"] 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 # ---- PIPELINE: five stages, each counts the rows it drops 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 try: f = int(k["index"]) - int(prev["index"]) except (TypeError, ValueError): dropped += 1 continue out.append(dict(k, m3=f)) return out, dropped def s4_summary(records): totals = {} for k in records: key = (k["region"], k["period"]) totals[key] = totals.get(key, 0) + k["m3"] return totals, 0 def s5_serve(summary): region = {} for (b, d), v in summary.items(): region.setdefault(b, []).append(v) return {b: round(sum(v) / len(v), 2) for b, v in region.items()}, 0 STAGES = [("screen", s1_screen), ("dedupe", s2_dedupe), ("consumption", s3_consumption), ("summary", s4_summary), ("serve", s5_serve)] def pipeline(records, stages=None): trace, data = [], records for name, f in (stages or STAGES): data, dropped = f(data) trace.append({"stage": name, "remaining": len(data), "dropped": dropped}) return data, trace # ---- INGESTION and REPROCESSING def ingest(records, load): """Ingestion stage: each load is given a stamp.""" return [dict(k, load=load) for k in records] def resend(records, periods, load): """The corrected load is resent ; the stamp changes, the value does not.""" return [dict(k, load=load) for k in records if k["period"] in periods] def store_append(store, new): """NON-idempotent write: the incoming row is appended to the end of the store.""" return store + new def store_keyed(store, new): """IDEMPOTENT write: the (subscriber , period) key is overwritten.""" table = {(k["subscriber"], k["period"]): k for k in store} for k in new: table[(k["subscriber"], k["period"])] = k return list(table.values()) def s2_dedupe_full(records): """FLAWED DEDUPLICATION: looks at the whole row, not at the key.""" seen, keep = set(), [] for k in records: stamp = tuple(sorted((a, str(v)) for a, v in k.items())) if stamp in seen: continue seen.add(stamp) keep.append(k) return keep, len(records) - len(keep) FLAWED = [("screen", s1_screen), ("dedupe", s2_dedupe_full), ("consumption", s3_consumption), ("summary", s4_summary), ("serve", s5_serve)] def c_uniqueness(data): return len(data) != len(set((k["subscriber"], k["period"]) for k in data)) def c_row_count(data, expected=2400, tolerance=0.05): return abs(len(data) - expected) / expected > tolerance ``` ## Duplicated Rows and an Unchanged Serving The second block applies five setups to the same source. The `raw` column gives the row count in the store, `dedupe` the rows dropped by that stage, `summary` the cell count of the fourth stage, and `north` one of the five served values. ```python SOURCE_RECORDS = generate_source() FIRST = ingest(SOURCE_RECORDS, 1) BASELINE, _ = pipeline(FIRST) def report(name, store, stages=None): serving, trace = pipeline(store, stages) print(f"{name:<30}{len(store):>6}{trace[1]['dropped']:>9}{trace[3]['remaining']:>9}" f"{serving['north']:>11.2f}{'same' if serving == BASELINE else 'changed':>9}") one_period = resend(SOURCE_RECORDS, {"2026-07"}, 2) full = resend(SOURCE_RECORDS, set(PERIODS), 2) print("baseline serving:", BASELINE) print("resent rows: one period", len(one_period), " full run", len(full)) print() print(f"{'setup':<30}{'raw':>6}{'dedupe':>9}{'summary':>9}{'north':>11}{'serving':>9}") report("single run", FIRST) report("appending store , one period", store_append(FIRST, one_period)) report("appending store , full run", store_append(FIRST, full)) report("keyed store , full run", store_keyed(FIRST, full)) report("flawed deduplication , full", store_append(FIRST, full), FLAWED) print() d = store_append(FIRST, full) print("uniqueness check:", c_uniqueness(d), " row count check:", c_row_count(d)) ``` ``` baseline serving: {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} resent rows: one period 200 full run 2400 setup raw dedupe summary north serving single run 2400 0 55 1212.55 same appending store , one period 2600 200 55 1212.55 same appending store , full run 4800 2400 55 1212.55 same keyed store , full run 2400 0 55 1212.55 same flawed deduplication , full 4800 0 55 2425.09 changed uniqueness check: True row count check: True ``` The second row is the headline of the lesson: when one period is backfilled, the appending store carries **2600** rows, which means **200 rows were duplicated**. The deduplication stage drops exactly those 200 rows, and the served value stays at **1212.55**. The third row shows the same thing at a larger scale: a full rerun takes the store to **4800** rows, deduplication drops **2400** rows, and the serving is again the same. This is the cleanest example of the third claim of the course. The size of the store has doubled, the pipeline has done the same work twice, yet **none of the five served values has moved**. An observation that looks at the output concludes that the pipeline is idempotent. But what is idempotent is not the pipeline, it is **a single stage inside the pipeline**: deduplication absorbs the surplus and tells nobody while doing so. The fourth row shows what real idempotence is. After the second run the keyed store still carries **2400** rows: duplicated rows **0**, rows dropped by deduplication **0**. The difference is not in the store but beyond the pipeline; while producing the right number, the appending store takes up **twice the space** and cannot be used without a deduplication stage. The last two rows are what the checks say. The uniqueness check fires on the appending store, and the row count check fires too — 4800 is far outside the five percent band around the expected 2400. So the detection lag of the corruption is **0**, and the check that catches it is **not a check that looks at the serving.** ## The Absorbed Correction So far the value of the resent row was the same as the first one's, so which one deduplication kept did not matter. In real backfills the situation is the reverse: the load is resent **because it was corrected**. The deduplication stage keeps the row it sees first, that is, **the old one**. In the setup below, the producer read the last period short, then corrected it and resent it. The corruption is deliberately placed in the last period: a shift in the middle of the series returns with the opposite sign in the following period and the two cancel out in the average. ```python def short_reading(records, period="2026-12", diff=100): """The producer read the last period short ; it will then correct it and resend.""" return [dict(k, index=k["index"] - diff) if k["period"] == period else k for k in records] SHORT = ingest(short_reading(SOURCE_RECORDS), 1) CORRECTION = resend(SOURCE_RECORDS, {"2026-12"}, 2) print("corrected rows:", len(CORRECTION)) print(f"{'setup':<32}{'raw':>6}{'dedupe':>9}{'north':>10}{'baseline':>10}") for name, store in (("short load , no correction", SHORT), ("appending store , correction", store_append(SHORT, CORRECTION)), ("keyed store , correction", store_keyed(SHORT, CORRECTION))): serving, trace = pipeline(store) print(f"{name:<32}{len(store):>6}{trace[1]['dropped']:>9}{serving['north']:>10.2f}" f"{BASELINE['north']:>10.2f}") ``` ``` corrected rows: 200 setup raw dedupe north baseline short load , no correction 2400 0 794.36 1212.55 appending store , correction 2600 200 794.36 1212.55 keyed store , correction 2400 0 1212.55 1212.55 ``` The second row is the most expensive finding of the lesson. The correction arrived at the pipeline, 200 rows were reprocessed, the store went up to 2600 rows, deduplication dropped 200 rows — and the served value stayed at **794.36**, that is, **exactly the same** as before the correction. The 200 rows that were dropped were the corrected ones. The deduplication stage absorbed the second run; here, absorbing is **losing**. This is the exact opposite reading of the previous section, and both come from the same behavior. When the resent row is the same as the first, deduplication saves the pipeline; when it is corrected, the same deduplication throws the correction away. The stage cannot tell the two apart, because all it has is the key and the key of the two rows is the same. The third row shows the solution: because the keyed store overwrites the old row with the new one, the serving returns to **1212.55**, that is, to the baseline. For a pipeline to count as idempotent, it is not enough that processing the same load twice leaves the result unchanged; **processing a corrected load a second time must also correct the result.** The appending store meets the first condition and does not meet the second. ## When the Key Is Chosen Wrongly The last row of the table opens up the most fragile assumption of the pipeline. The deduplication stage looks at the `(subscriber, period)` key. Had the same stage been written to look at **the whole row**, the resent row would look different because of the load stamp and would not be eliminated. This is not a made-up flaw: asking whether a load is entirely the same seems intuitively safer, and that is exactly why it gets written. ```python appending, keyed = FIRST, FIRST print(f"{'run':<6}{'appending raw':>15}{'north':>9}{'flawed north':>15}" f"{'keyed raw':>18}{'north':>9}") for run in range(1, 4): if run > 1: new = resend(SOURCE_RECORDS, set(PERIODS), run) appending = store_append(appending, new) keyed = store_keyed(keyed, new) serve_append, _ = pipeline(appending) serve_flawed, _ = pipeline(appending, FLAWED) serve_keyed, _ = pipeline(keyed) print(f"{run:<6}{len(appending):>15}{serve_append['north']:>9.2f}" f"{serve_flawed['north']:>15.2f}{len(keyed):>18}" f"{serve_keyed['north']:>9.2f}") print() summ, _ = s4_summary(s3_consumption(s2_dedupe_full(s1_screen(appending)[0])[0])[0]) summ_base, _ = s4_summary(s3_consumption(s2_dedupe(s1_screen(FIRST)[0])[0])[0]) print("summary ('north', '2026-07') baseline:", summ_base[("north", "2026-07")], " flawed deduplication:", summ[("north", "2026-07")]) print("keys kept by deduplication:", len(set((k["subscriber"], k["period"]) for k in FIRST))) ``` ``` run appending raw north flawed north keyed raw north 1 2400 1212.55 1212.55 2400 1212.55 2 4800 1212.55 2425.09 2400 1212.55 3 7200 1212.55 3637.64 2400 1212.55 summary ('north', '2026-07') baseline: 1251 flawed deduplication: 3753 keys kept by deduplication: 2400 ``` The three-run table sets two sentences side by side. In the pipeline that works with the right key, the raw store goes from 2400 to 4800 to 7200 and the served value is **1212.55 in all three runs**. In the pipeline that works with the flawed key, the same raw store and the same reruns make the served value grow as **1212.55 → 2425.09 → 3637.64**: in direct proportion to the number of runs. The same shows in the summary layer: the `2026-07` cell of the north is 1251 at the baseline and **3753** after the third run. The only difference between the two pipelines is the key choice of one stage. No corruption class, no field name changed, no type drifted; the only thing that happened on the producer side is that the same correct load was sent a second time. This is the form that the second claim of the course takes in this lesson: **a pipeline's own stage, with a flawed key, can triple the served number without any source corruption.** The column of the keyed store says where the cost lies. Idempotence is not free: at write time a key table of **2400 entries** is kept and every row is looked up before it is written. The deduplication stage carries a set of the same size. In return, the store stays at 2400 rows regardless of the number of runs, and the correctness of the pipeline no longer depends on the key of a single stage. ## Summary - Reprocessing is an ordinary job; in an appending store, backfilling one period takes the raw rows from 2400 to 2600, a full rerun to 4800, and the third run to 7200. - None of the five served values shows this growth: the north is 1212.55 in all three runs, because the deduplication stage absorbs the surplus and does not say so while absorbing. - What is idempotent is not the pipeline but a single stage inside it; when keyed writing is used, duplicated rows fall to 0 and the store stays independent of the number of runs. - When the resent load carries a **correction**, the same deduplication throws it away: the served value stays at 794.36, whereas the keyed store returns to 1212.55 with the same correction. Idempotence is not that the second run leaves the result unchanged, but **that it corrects when it should correct**. - When deduplication looks at the whole row instead of `(subscriber, period)`, the same reruns carry the served value from 1212.55 to 2425.09 and then to 3637.64; the summary cell goes from 1251 to 3753. - The checks that catch duplication are uniqueness and row count, and both fire at stage 0; no check that looks at the serving can see it. ## Next Step In three lessons the check always looked at the same place: **the row itself**. Does it have its fields, is its type right, is its key unique, is its value in line with the previous periods. These questions give no answer when a serving value is corrupted, because none of them says **where the row came from**. Someone looking at the figure 3753 produced by the flawed deduplication has to read the whole pipeline from the beginning to find which loads contributed to that cell and at which link of the chain the error is. The next lesson counts this search: how many stages and how many rows one has to look at to go from a corrupted serving value back to the source, with a record kept and without one.