Training Pipelines

The first six lessons of this course made the model reproducible: the experiment run was recorded, the data and the model were versioned, the feature path in training was compared with the path in serving, the environment items were pinned and the stage transitions were tied to a registry. Their shared assumption can be written in one sentence: the model has been trained once, and the problem is being able to train it again. None of them asked how the trained model would reach production.

This lesson opens the first step of that question. In five stages, the pipeline of the Introduction to Data Engineering course went from 2400 rows to 2400, to 2200, to 55 summary cells and to 5 serving values. The model is not attached to the end of that chain; it is attached to the output of the third stage, that is, to the 2200 consumption rows. The question of this lesson is what becomes measurable once the steps that follow that point stop being a sequence run by hand and become an end-to-end runnable training pipeline.

• TD1. The source, the pipeline and the model are fictional; no real machine learning library, experiment tracking setup, feature store or serving framework is called. The seed is 20260218.
• TD2. In this course the training pipeline is treated on a single machine; cluster sizing and distribution were measured in the Distributed Data Processing course and are not repeated.
• TD3. Linking the steps to each other, solving the graph and the retry procedure were measured in the Data Pipelines and Orchestration course. The question asked here is not the order but what each step does to the rows.
• TD4. The label: a row whose consumption exceeds 40 m3 counts as “high”. The distribution is imbalanced and is written in every lesson.
• TD5. The training periods are 2026-02–2026-09, the test periods are 2026-10–2026-12.
• TD6. The model is a threshold with three weights; training is a search on a grid of 27 candidates.
• TD7. The features are computed once and are independent of the weights.
• TD8. The baseline is the steps run by hand: the intermediate outputs stay on hand, and a re-run is partial.
• TD9. Resolution: the test set is 600 rows, and the smallest step of accuracy is 0.0017.
• TD10. Time is not written as raw milliseconds; the measures are rows, steps and periods.

The Six Steps of the Training Pipeline

The first block sets up the source, the first three stages of the pipeline and the model side. None of it is specific to this lesson: the first three come from the data pipeline, and the model side is the shared definition of this course.

# FICTIONAL. Meter readings, pipeline and model are defined inside the lesson; no real
# machine learning library, experiment tracking setup or serving framework 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


# ---- The first three stages of the data pipeline
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):
    """Consumption from the period difference. Without a previous period the row cannot be computed."""
    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


# ---- The model side
TRAIN, TEST, THRESHOLD = PERIODS[1:9], PERIODS[9:], 40


def label(k):
    return 1 if k["m3"] > THRESHOLD else 0


def features(records, window=None, digits=4):
    """window=None -> full history. Independent of the weights , computed once."""
    ordered = sorted(records, key=lambda k: PERIODS.index(k["period"]))
    sub, reg = {}, {}
    for k in ordered:
        sub.setdefault(k["subscriber"], []).append((PERIODS.index(k["period"]), k["m3"]))
        reg.setdefault(k["region"], []).append((PERIODS.index(k["period"]), k["m3"]))

    def mean(series, i):
        chosen = [v for d, v in series if (d < i and (window is None or d >= i - window))]
        return round(sum(chosen) / len(chosen), digits) if chosen else 0.0

    return {(k["subscriber"], k["period"]): {
        "subscriber_mean": mean(sub[k["subscriber"]], PERIODS.index(k["period"])),
        "region_mean": mean(reg[k["region"]], PERIODS.index(k["period"])),
        "commercial": 1 if k["type"] == "commercial" else 0} for k in records}


GRID = [(a, b, c) for a in (0.6, 0.8, 1.0) for b in (0.0, 0.2, 0.4) for c in (0, 4, 8)]


def score(o, w):
    return w[0] * o["subscriber_mean"] + w[1] * o["region_mean"] + w[2] * o["commercial"]


def train(records, feats):
    subset = [k for k in records if k["period"] in TRAIN]
    best, best_w = -1.0, None
    for w in GRID:
        d = sum(1 for k in subset
                if (1 if score(feats[(k["subscriber"], k["period"])], w) > THRESHOLD else 0) == label(k))
        if d / len(subset) > best:
            best, best_w = d / len(subset), w
    return {"weights": best_w, "train_accuracy": round(best, 4),
            "candidates": len(GRID), "train_rows": len(subset)}


def evaluate(records, model, feats):
    subset = [k for k in records if k["period"] in TEST]
    d = sum(1 for k in subset
            if (1 if score(feats[(k["subscriber"], k["period"])], model["weights"]) > THRESHOLD else 0)
            == label(k))
    return round(d / len(subset), 4)

The second block sets up the pipeline. The first three of the six steps are the stages above; the remaining three are labeling, feature generation and the split. run_pipeline reports the input of each step, its output and its own dropped-row counter separately; when a start value is given, the first steps come from the stored output and only the rest re-runs.

def t4_label(data):
    return [dict(k, y=label(k)) for k in data], 0


def t5_features(data):
    feats = features(data)
    return [dict(k, **feats[(k["subscriber"], k["period"])]) for k in data], 0


def t6_split(data):
    return {"train": [k for k in data if k["period"] in TRAIN],
            "test": [k for k in data if k["period"] in TEST]}, 0


STEPS = [("screen", s1_screen), ("dedupe", s2_dedupe), ("consumption", s3_consumption),
         ("label", t4_label), ("features", t5_features), ("split", t6_split)]


def size(data):
    return len(data) if isinstance(data, list) else sum(len(x) for x in data.values())


def run_pipeline(source, start=0, stored=None):
    """If start=p the first p steps come from the stored output , the rest re-run."""
    data = source if start == 0 else stored[STEPS[start - 1][0]]
    inter, trace, processed = dict(stored or {}), [], 0
    for i, (name, f) in enumerate(STEPS):
        if i < start:
            continue
        n_in = size(data)
        data, counter = f(data)
        inter[name], processed = data, processed + n_in
        trace.append((name, n_in, size(data), counter))
    return data, trace, inter, processed


def assess(split_data):
    subset = split_data["train"] + split_data["test"]
    feats = {(k["subscriber"], k["period"]): {a: k[a] for a in ("subscriber_mean", "region_mean", "commercial")}
             for k in subset}
    m = train(subset, feats)
    return m, evaluate(subset, m, feats), feats


SOURCE = generate_source()
SPLIT, TRACE, INTER, PROCESSED = run_pipeline(SOURCE)
print(f"{'step':<11}{'input':>7}{'output':>7}{'lost':>10}{'step counter':>15}")
for name, n_in, n_out, counter in TRACE:
    print(f"{name:<11}{n_in:>7}{n_out:>7}{n_in - n_out:>10}{counter:>15}")
print("split:", {a: len(b) for a, b in SPLIT.items()},
      " history-less rows:", sum(1 for k in INTER["features"] if k["subscriber_mean"] == 0.0))
DISTRIBUTION = {0: 0, 1: 0}
for k in INTER["label"]:
    DISTRIBUTION[k["y"]] += 1
print("label distribution:", DISTRIBUTION, " base class ratio:",
      round(DISTRIBUTION[0] / len(INTER["label"]), 4), " processed rows:", PROCESSED)

step         input output      lost   step counter
screen        2400   2400         0              0
dedupe        2400   2400         0              0
consumption   2400   2200       200              0
label         2200   2200         0              0
features      2200   2200         0              0
split         2200   2200         0              0
split: {'train': 1600, 'test': 600}  history-less rows: 200
label distribution: {0: 1752, 1: 448}  base class ratio: 0.7964  processed rows: 13800

The last two columns of the table contradict each other, and the contradiction sits in a single step. The input of the consumption step is 2400 and its output is 2200; 200 rows are lost in between. The step’s own counter reads 0. Both are correct: the counter counts only the unexpected drop, that is, the row whose previous period cannot be found. The 200 rows of the first period drop by definition, because consumption is the difference between two periods. Looking at the dropped-row counter of a step does not tell how many rows that step lost; the difference between the input and the output has to be measured separately.

The second contradiction is in the feature step and runs the other way. This step drops no rows at all, but 200 rows in its output have a subscriber mean of 0.0, because the 2026-02 rows have no period before themselves. A step whose dropped rows are zero can still corrupt the data; the counter of the pipeline does not see it.

The split step really does drop no rows: 1600 training rows and 600 test rows add up to 2200. An end-to-end run processes 13800 rows across six steps.

The Number the Evaluation Step Produces

The seventh and eighth steps of the pipeline are training and evaluation. The third block runs both and puts the resulting number next to the base class ratio.

MODEL, ACCURACY, FEAT_TABLE = assess(SPLIT)
TEST_SET = SPLIT["test"]
PREDICTIONS = [1 if score(FEAT_TABLE[(k["subscriber"], k["period"])], MODEL["weights"]) > THRESHOLD else 0
               for k in TEST_SET]
BASE_RATIO_TEST = sum(1 for k in TEST_SET if k["y"] == 0) / len(TEST_SET)
print("weights:", MODEL["weights"], " candidates:", MODEL["candidates"],
      " training rows:", MODEL["train_rows"], " training accuracy:", MODEL["train_accuracy"])
print(f"test accuracy: {ACCURACY:.4f}  test base class ratio: {BASE_RATIO_TEST:.4f}"
      f"  gain over base class ratio: {ACCURACY - BASE_RATIO_TEST:.4f}")
print("test prediction distribution:", {0: PREDICTIONS.count(0), 1: PREDICTIONS.count(1)})

weights: (0.6, 0.4, 0)  candidates: 27  training rows: 1600  training accuracy: 0.7931
test accuracy: 0.8067  test base class ratio: 0.8067  gain over base class ratio: 0.0000
test prediction distribution: {0: 600, 1: 0}

The number the pipeline produces is 0.8067, and read on its own it looks good. The two numbers next to it make it unreadable. The label distribution is 1752 to 448, that is, the base class ratio is 0.7964; in the test set this ratio is 0.8067. The accuracy of the model is exactly the base class ratio, and the gain over the base class ratio is 0.0000.

The last line writes the reason: the model says 0 to 600 of the 600 test rows. It counts no row as “high”, so its accuracy is nothing but the share of the majority class. The resolution of the test set is 0.0017; the gain is not even below this smallest step, it is exactly zero. If the evaluation step of a training pipeline prints a single accuracy, that pipeline is not measuring.

Re-Running End to End

The source is not something fixed. In the period 2026-07 the readings of 40 subscribers arrive late and are backfilled afterwards. The fourth block first runs with the incomplete source, and then resumes the pipeline from different points with the backfilled source. The stale step count is the number of steps that come from the stored old output.

LATE = [k for k in SOURCE if not (k["period"] == "2026-07" and k["subscriber"] <= "A040")]
splits_late, trace_late, inter_late, _ = run_pipeline(LATE)
model_late, test_late, _ = assess(splits_late)
print("late-arriving source:", len(LATE), " consumption output:", trace_late[2][2],
      " training rows:", model_late["train_rows"], f" test accuracy: {test_late:.4f}")
print()
print(f"{'stale step':>11}{'steps rerun':>14}{'processed rows':>15}"
      f"{'training rows':>14}{'training accuracy':>18}{'test accuracy':>15}")
for p in range(7):
    if p == 6:
        parts, processed = inter_late["split"], 0
    else:
        parts, _, _, processed = run_pipeline(SOURCE, start=p, stored=inter_late)
    m, s, _ = assess(parts)
    print(f"{p:>11}{6 - p:>14}{processed:>15}{m['train_rows']:>14}"
          f"{m['train_accuracy']:>18.4f}{s:>15.4f}")

late-arriving source: 2360  consumption output: 2120  training rows: 1520  test accuracy: 0.8067

 stale step   steps rerun processed rows training rows training accuracy  test accuracy
          0             6          13800          1600            0.7931         0.8067
          1             5          11080          1520            0.7914         0.8067
          2             4           8720          1520            0.7914         0.8067
          3             3           6360          1520            0.7914         0.8067
          4             2           4240          1520            0.7914         0.8067
          5             1           2120          1520            0.7914         0.8067
          6             0              0          1520            0.7914         0.8067

The 40 late-arriving rows take 80 rows away from the output of the pipeline: they drop themselves, and 40 rows of the next period drop because they cannot find their previous period. The training set is built with 1520 rows instead of 1600.

The table says a single thing. After the backfill, the only setup that raises the training rows to 1600 is the pipeline whose stale step is zero, that is, the pipeline that re-runs end to end from the source. It is enough for even a single step to come from the stored old output: even if the pipeline truly re-runs the remaining five steps, the training rows stay at 1520, because the lost rows were already missing from the output of the first step. A partial run saves by bringing the processed rows down from 13800 to 11080, and in return it never sees the corrected source.

The reported number is 0.8067 in all seven setups. No one who looks at the test accuracy can tell a model trained with 1600 rows from a model trained with 1520 rows. The only place that shows the difference is the training accuracy, and there the difference is 0.0017. The rule of the course finds its full counterpart here: a number that cannot be reproduced is not a measurement but a memory, and a pipeline’s being runnable end to end means that the source the number came from is determinable.

The cost of an end-to-end run is also in the table: every run reprocesses 13800 rows; in the fully stale setup this number is 0. Pinning is not free.

Summary

• The training pipeline branches off not from the fifth stage of the data pipeline but from the third: the model is attached not to the 5 serving values but to the 2200 consumption rows.
• When the input and the output of the six steps are counted separately, the consumption step loses 200 rows, but the step’s own counter reads 0; the counter counts only the unexpected drop.
• The feature step leaves 200 rows history-less without dropping any row; a step whose dropped rows are zero can still corrupt the data.
• The label distribution is 1752 to 448, the base class ratio is 0.7964, and in the test set it is 0.8067. The model says 0 to 600 of the 600 rows; its accuracy is 0.8067 and its gain over the base class ratio is 0.0000.
• The 40 late-arriving rows take 80 rows away from the training set; after the backfill, the only setup that returns to 1600 rows is the end-to-end run whose stale step is zero, and its cost is 13800 processed rows per run.
• All seven setups report the same 0.8067; the test accuracy cannot tell training with 1520 rows from training with 1600 rows.

Next Step

Once the pipeline of this lesson was set up, it always ran the same way, and every time it ran it produced a model. Whether the produced model would go to release was never asked. The model of 1520 rows trained with the missing late-arriving rows and the backfilled model of 1600 rows could have gone to release with equal ease, because both reported the same number. The next lesson puts gates between the run and the release: a code gate, a data gate and a model gate. The measurement is how the six corruption classes defined in the Introduction to Data Engineering course are distributed over these three gates, and which class is the one no gate stops.