--- title: 'Training Pipelines' source: 'https://academia.sh/en/courses/mlops-fundamentals/training-pipelines' course: 'MLOps Fundamentals' language: en updated: '2026-10-21T05:03:02+00:00' license: 'CC BY-SA 4.0' objectives: - 'Builds a six-step training pipeline from the output of the data pipeline to training and evaluation with working code.' - "Measures the difference between the input and the output of each step and the step's own dropped-row counter separately, and names the step where the two contradict." - "Reads the test accuracy next to the base class ratio and counts the model's gain over the base class ratio." - 'Compares the training rows produced by a partial run and by an end-to-end run when the source is backfilled.' --- # Training Pipelines Counting the training pipeline as the first step of a trained model's path to production: the rows dropped by each of six steps are measured separately, and while the consumption step loses 200 rows its own counter reads 0, the feature step leaves 200 rows history-less without dropping a single row, the split separates 1600 and 600 without dropping a single row; with a label distribution of 1752 to 448 and a base class ratio of 0.7964, the model says 0 to 600 of the 600 rows in the test set and its accuracy of 0.8067 is exactly the base class ratio; when the 40 late-arriving rows are backfilled, the only setup that raises the training rows from 1520 to 1600 is the pipeline that re-runs end to end, and all seven setups report the same 0.8067. 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. ```python # 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. ```python 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. ```python 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. ```python 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**.