--- title: 'Ingestion Types' source: 'https://academia.sh/en/courses/data-pipelines/ingestion-types' course: 'Data Pipelines and Orchestration' language: en updated: '2026-10-21T05:02:58+00:00' license: 'CC BY-SA 4.0' objectives: - 'Builds the pipeline as a directed acyclic graph of four tasks per period and counts its nodes and edges.' - 'Compares full load, incremental load, and change data capture by the rows they transport and the changes they miss.' - 'Shows in which method a late-arriving row and a master data correction stay invisible.' - 'Calculates which check catches the missed change and after how many stages.' --- # Ingestion Types Counting side by side what three ingestion methods transport and what each misses: a graph built from four tasks per period carries 48 nodes and 47 edges, full load transports 15,560 rows over twelve runs and misses no change, incremental load transports 2,360 rows and misses 148 changes, and change data capture transports 2,508 rows and misses none. The 40 late-arriving rows and 108 uncorrected rows that incremental load misses pull the serving down in every region, yet the row count check stays silent at a ratio of 0.0167 and only the distribution check catches the corruption, at 0.3469 in stage 4. The previous course measured, across twelve lessons, **where** data sits and in **what layout**: partitioning, file layout, compression, and cells read were counted side by side. **Who fills** that layout, **when**, and **in what order** was never asked. A pipeline does not fill itself; every period someone triggers it, the steps run in a sequence, and what the rest should do when one of them drops out of that sequence has to be written down in advance. This course measures the pipeline through the **run**, and its rule is this: a pipeline's number is not that it ran correctly once but that it **gives the same result when run again**; a pipeline that cannot be rerun counts as **unmeasured**. In every lesson three numbers sit side by side: the unorchestrated baseline, the orchestrated setup, and the cost. This lesson builds the trio and the pipeline's task graph, then asks the first question: by **which method** is data ingested from the source, and what does that method **miss**. - **DI1.** The code is **fictional**; no real store, engine, messaging system, or orchestrator is called. The source is the fictional meter readings of a measurement network: 200 subscribers, 12 periods, 2400 rows, seed 20260218. - **DI2.** The pipeline is deterministic; no randomness library is used and all numbers are independent of the run. - **DI3.** The baseline is the **unorchestrated run**: a single manual trigger, no retries, no atomic write, no backfill. - **DI4.** The task graph carries four **tasks** per period: `ingest`, `screen`, `consumption`, `summary`. `consumption` also depends on the `screen` task of the previous period. - **DI5.** A late-arriving row is the `late_arrival` corruption of the Introduction to Data Engineering course: 40 rows of the 2026-07 period are absent from that period's run and arrive before the next run. It is not an attack but a producer-side accident. - **DI6.** The master data correction fixes the region of 12 subscribers before the tenth run, and the correction is applied to **all periods** of those subscribers. - **DI7.** The target is keyed: the `(subscriber, period)` pair identifies a row, and when the same key is written again it is overwritten. The write is therefore **idempotent**. - **DI8.** Transported rows are the sum of the number of rows the `ingest` task passes into the pipeline over twelve runs. The step of the row ratio is 1/2400, that is 0.0004; no ratio is printed on the serving. ## The Task Graph of the Pipeline A pipeline is a **directed acyclic graph** of interdependent tasks. The graph concept and topological ordering were measured in the Graphs topic of the Data Structures course; the algorithm is not repeated here, the graph is **built and used**. There are four tasks per period and the one real subtlety is this: consumption is a **difference** computation, so the `consumption` task depends on the `screen` task of its own period as well as on the `screen` task of the **previous period**. The chain reaching across periods explains why, in later lessons, a whole queue waits when one task fails. ```python # FICTIONAL. No real store, engine, or orchestrator is called; the meter # readings of the measurement network are fictional, 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 = ["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(): # one reading per subscriber per period 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 stage 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): # per-region mean over the periods 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 def read_result(records): v, _ = s1_screen(records) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) summ, _ = s4_summary(v) serving, _ = s5_serve(summ) return serving, len(v), summ # ---- FICTIONAL task graph: four tasks per period. TASKS = ("ingest", "screen", "consumption", "summary") DEPENDENCIES = {"ingest": (), "screen": ("ingest",), "consumption": ("screen",), "summary": ("consumption",)} def graph(): """`consumption` also depends on the previous period's `screen`.""" nodes, edges = [], [] for p in PERIODS: for t in TASKS: nodes.append((t, p)) for d in DEPENDENCIES[t]: edges.append(((d, p), (t, p))) i = PERIODS.index(p) if i > 0: edges.append((("screen", PERIODS[i - 1]), ("consumption", p))) return nodes, edges def order_tasks(nodes, edges): incoming = {u: 0 for u in nodes} successors = {u: [] for u in nodes} for a, b in edges: incoming[b] += 1 successors[a].append(b) ready = [u for u in nodes if incoming[u] == 0] order = [] while ready: u = ready.pop(0) order.append(u) for v in successors[u]: incoming[v] -= 1 if incoming[v] == 0: ready.append(v) return order nodes, edges = graph() order = order_tasks(nodes, edges) raw = generate_source() serving, n, _ = read_result(raw) print("graph", len(nodes), "nodes", len(edges), "edges , order", len(order), "tasks") print("first three tasks", order[:3]) print("source", len(raw), "rows , consumption", n, "rows") print("healthy serving", serving) ``` ``` graph 48 nodes 47 edges , order 48 tasks first three tasks [('ingest', '2026-01'), ('ingest', '2026-02'), ('ingest', '2026-03')] source 2400 rows , consumption 2200 rows healthy serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} ``` Forty-eight nodes, forty-seven edges, and a graph that can be fully ordered. The edge count being one less than the node count says the graph is a **forest in one piece**: apart from the `ingest` tasks of the twelve periods, every node has exactly one predecessor. What this lesson measures is not the shape of the graph but what the very first node does: **what the `ingest` task asks of the source**. ## Three Ingestion Methods **Full load** reads the whole of the source as it is at that moment on every run and rewrites the target from scratch. **Incremental load** asks only for the periods after the last ingested period; the boundary is a period number and advances on every run. **Change data capture** reads the source's own change stream: it transports insert and update events and never transports a row that was not touched. The block below models the source over time and runs the three methods over the same twelve runs. Two producer-side events fall on the source: forty rows of 2026-07 arrive late, and the region of the twelfth subscriber is corrected before the tenth run. ```python # ---- FICTIONAL source system over time. Two producer-side events: # 1) The `late_arrival` corruption of the Introduction to Data Engineering # course: 40 rows of 2026-07 are absent in that period's run and arrive # before the next run. # 2) Master data correction: the region of 12 subscribers was entered wrongly # and is corrected for all of their periods before run 10. LATE = [f"A{i:03d}" for i in range(1, 41)] WRONG = [f"A{i:03d}" for i in range(101, 113)] def source_snapshot(run): """The state of the source as visible at the moment of run number `run`.""" records = [] for k in generate_source(): i = PERIODS.index(k["period"]) + 1 if i > run or (i == 7 == run and k["subscriber"] in LATE): continue y = dict(k) if k["subscriber"] in WRONG and run < 10: y["region"] = REGION[(REGION.index(k["region"]) + 1) % 5] records.append(y) return records def full_load(): """On every run the whole source is reread , the target is wiped and written.""" target, transported = {}, 0 for run in range(1, 13): snapshot = source_snapshot(run) transported += len(snapshot) target = {(k["subscriber"], k["period"]): k for k in snapshot} return list(target.values()), transported def incremental_load(): """Only the periods after the last ingested period are read.""" target, transported, boundary = {}, 0, 0 for run in range(1, 13): fresh = [k for k in source_snapshot(run) if PERIODS.index(k["period"]) + 1 > boundary] transported += len(fresh) for k in fresh: target[(k["subscriber"], k["period"])] = k boundary = run return list(target.values()), transported def change_data_capture(): """The source's change stream is read , insert and update events are transported.""" target, transported, prev = {}, 0, {} for run in range(1, 13): snapshot = {(k["subscriber"], k["period"]): k for k in source_snapshot(run)} events = [k for key, k in snapshot.items() if prev.get(key) != k] transported += len(events) for k in events: target[(k["subscriber"], k["period"])] = k prev = snapshot return list(target.values()), transported actual = {(k["subscriber"], k["period"]): k for k in raw} correct_serving, _, correct_summary = read_result(raw) print("method transported target missing wrong consumption row ratio distribution width") for name, f in (("full", full_load), ("incremental", incremental_load), ("change capture", change_data_capture)): tgt, transported = f() tm = {(k["subscriber"], k["period"]): k for k in tgt} missing = len(set(actual) - set(tm)) wrong = sum(1 for a in tm if actual[a] != tm[a]) s, t, summ = read_result(tgt) width = max(abs(v - correct_summary[a]) / max(abs(correct_summary[a]), 1) for a, v in summ.items()) print(f"{name:<14} {transported:>11} {len(tm):>6} {missing:>7} {wrong:>5} {t:>11} " f"{abs(len(tm) - 2400) / 2400:>9.4f} {width:>18.4f}") print() for name, f in (("full", full_load), ("incremental", incremental_load), ("change capture", change_data_capture)): s, _, _ = read_result(f()[0]) print(f"{name:<14}", {b: round(s.get(b, 0) - correct_serving[b], 2) for b in correct_serving}) ``` ``` method transported target missing wrong consumption row ratio distribution width full 15560 2400 0 0 2200 0.0000 0.0000 incremental 2360 2360 40 108 2120 0.0167 0.3469 change capture 2508 2400 0 0 2200 0.0000 0.0000 full {'central': 0.0, 'north': 0.0, 'east': 0.0, 'west': 0.0, 'south': 0.0} incremental {'central': -11.18, 'north': -28.1, 'east': -39.09, 'west': -36.91, 'south': -70.0} change capture {'central': 0.0, 'north': 0.0, 'east': 0.0, 'west': 0.0, 'south': 0.0} ``` ## Transported Rows versus Missed Changes The first column of the table writes the cost of full load: over twelve runs **15,560 rows** were transported, whereas the rows that end up in the target are **2400**. The price of transporting the same data six and a half times is that it misses no change. The forty late rows settle into place on their own in the eighth run, and the region correction spreads to all periods in the tenth run, because every run rereads the source. Full load has no flaw; it is only expensive, and **its expense grows linearly as the source grows**. Incremental load transports the fewest rows: **2360**. Its cost sits on the same row. The seventh run sees only 160 rows of the period; the forty rows arrive later, but the boundary is now at seven and those rows are never requested again. The region correction likewise lands only on the periods after the tenth run: nine periods of twelve subscribers, that is **108 rows**, stay with their old region in the target. In total **148 changes are missed**, of which 40 are missing rows and 108 are wrong rows. Incremental load does not **know** what it missed: it has no mechanism that would notice that something behind the boundary changed. Change data capture sees both events, and the rows it transports for this are **2508**. That is only **148 rows** more than incremental load and **13,052 rows** fewer than full load. The difference is exactly the number of missed changes, and this is no coincidence: what change data capture transports is **the change itself**. Its cost is not in rows but at the source; the source has to publish its change stream, and for a source that does not publish it the method cannot be set up at all. ## Which Check Sees the Missed Change Incremental load's target pulls the serving down in every region: the smallest deviation is **−11.18** in central, the largest **−70.00** in south. Is there anyone who would see this deviation? The check set built in the Introduction to Data Engineering course gives two different answers to this question. The ratio of the row count check is **0.0167**; because the default tolerance is 0.05, the check **stays green**. The schema, type, and uniqueness checks are green as well, because the schema of the missing rows is not broken, they are merely absent. The region set is also complete: the region of twelve subscribers is wrong, but the wrong region is on the list too. The only check that catches it is the **distribution check**, and at the summary layer the largest cell deviation of **0.3469** exceeds the 0.25 threshold. This check runs at stage 4; that is, the corruption becomes visible only after passing through ingestion, screening, deduplication, and the consumption computation. The detection lag is **4 stages**, and by then the 2120-row intermediate table has already been written. Method choice turns here into a choice of measure: incremental load cuts transported rows by a factor of 6.6, misses 148 changes in return, and this leak is reported only by the most expensive check, at the latest stage. ## Summary - The pipeline is a directed acyclic graph built from four tasks per period: 48 nodes, 47 edges, and a run order that can be fully sorted. `consumption` depends on the `screen` task of the previous period, so the chain reaches across periods. - Full load misses no change and pays its cost in transported rows: for a target of 2400 rows, 15,560 rows are transported over twelve runs. - Incremental load transports the fewest rows (2360) but cannot see what happens behind the boundary: 40 late-arriving rows and 108 uncorrected rows, 148 changes in total, are missed. - Change data capture sees both events and transports only as many rows more than incremental load as the number of missed changes, 148; its condition is that the source publishes its change stream. - The row count check does not see the missed change, at a ratio of 0.0167; only the distribution check reports it, at 0.3469 and with a lag of 4 stages. ## Next Step In this lesson the `ingest` task read the source in a **single move**: `source_snapshot` was called and the rows were in hand. Source systems are not read like this. An endpoint serves data page by page, limits how many calls can be made per window, and some of the calls drop temporarily. The next lesson puts the same `ingest` task under these constraints and counts two things: **the calls completed and the rows ingested under the constraint**. The trade-off to be measured is this: raising the page size lowers the call count, but the rows missed by a single dropped call grow by the same ratio.