Ingestion Types

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.

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

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