--- title: 'Continuous Integration for Machine Learning' source: 'https://academia.sh/en/courses/mlops-fundamentals/continuous-integration-for-machine-learning' course: 'MLOps Fundamentals' language: en updated: '2026-10-21T05:03:02+00:00' license: 'CC BY-SA 4.0' objectives: - 'Writes a working procedure that places the code, data and model gates between the run and the release.' - 'Measures at which gate and at which step each of the six corruption classes stops, and names the class that no gate stops.' - 'Shows that a corruption stops exactly at the tolerance of a gate, and counts whether an intact run stops when the tolerance is changed.' - 'Computes the cost of the gates from the runs stopped and the model lost when they stop.' --- # Continuous Integration for Machine Learning Counting which corruption each of the code, data and model gates placed between the run and the release stops: of the six corruption classes the data gate stops four and the model gate stops three, the code gate stops none but catches a feature code change at the first step without looking at production data at all; the unit change stops only at the model gate, by dropping the training accuracy from 0.7931 to 0.5919; the late-arriving row stops at no gate because its deviation in training rows is exactly 0.0500 and the tolerance is 0.0500 with a strict comparison; two of the four runs the data gate stopped would have produced exactly the same model as the reference, and in eight of eight runs the test accuracy of 0.8067 cannot rise above the base class ratio. The pipeline of the previous lesson produced a model every time it ran, and whether the produced model **would go to release** was never asked. The model trained with 1520 rows because of the missing late-arriving rows and the model trained with the backfilled 1600 rows both reported the same **0.8067**; both could have gone to release with equal ease. This lesson places **gates** between the run and the release. A gate is a check that stops a run before it reaches release. There are three, and each looks at a different thing: the **code gate** looks at the code that was written, the **data gate** at the input of the pipeline, the **model gate** at the numbers of the trained model. The question itself is one sentence: how do the six corruption classes defined in the Introduction to Data Engineering course distribute over these three gates, and which class is the one **no gate stops**. - **TD11.** The source, the pipeline, the model and the gates are **fictional**; no real machine learning library or continuous integration product is called. The seed is **20260218**. - **TD12.** The procedure of continuous integration itself, its triggers, the ordering of its stages and its deployment forms were measured in the Continuous Integration and Delivery course of the DevOps and Platform Engineering curriculum, and pipeline orchestration in the Data Pipelines and Orchestration course. They are **not repeated**; this lesson adds only the **data and model gates**. - **TD13.** None of the six corruption classes is an attack; all are accidents on the producer side and each corrupts **a single period** (`2026-07`). - **TD14.** The gates run in order: code **1**, data **2**, model **3**. The number of the gate that first stops an event also says how much the corruption has dirtied the pipeline. - **TD15.** The code gate **never sees production data**; it runs on a fixed input of three rows. - **TD16.** The **reference** that the model gate compares against is the model run with the intact source; the tolerances are **0.0500** for training rows and **0.0100** for training accuracy. - **TD17.** The label distribution is **imbalanced**; the base class ratio is written in this lesson too. - **TD18.** The baseline is the **setup without gates**: there is no gate, every run goes to release. - **TD19.** Resolution: the test set is **600 rows**, the step of accuracy is **0.0017**; the source is **2400 rows**, the step of the row ratio is **0.0004**. - **TD20.** Time is not written as raw milliseconds; the measures are **runs**, **rows** and **steps**. ## The Three Gates The first block sets up the source, the six corruption classes, the first three stages of the pipeline and the model. None of it is specific to this lesson. ```python # FICTIONAL. Source, pipeline, model and gates are defined inside the lesson; no real # machine learning library or continuous integration product 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 # ---- Six corruption classes. None of them is an attack , all are producer-side accidents. CORRUPTIONS = ["late_arrival", "duplicated", "schema_changed", "unit_changed", "missing_partition", "type_drift"] def corrupt(records, cls, period="2026-07"): """Corrupts a single period , the remaining eleven periods stay intact.""" out = [] for k in records: if k["period"] != period: out.append(k) elif cls == "late_arrival": if k["subscriber"] > "A040": # 40 subscribers never arrived this period out.append(k) elif cls == "duplicated": out.append(k) if k["subscriber"] <= "A060": # 60 records arrived twice out.append(dict(k)) elif cls == "schema_changed": y = dict(k) y["index_value"] = y.pop("index") # field name changed out.append(y) elif cls == "unit_changed": out.append(dict(k, index=k["index"] * 1000, unit="m3")) # label did not change elif cls == "missing_partition": if k["region"] != "north": # one region's partition was never written out.append(k) elif cls == "type_drift": out.append(dict(k, index=str(k["index"]))) # number turned into text return out 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 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): 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), "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), round(sum(1 for k in subset if label(k) == 0) / len(subset), 4) def run(records, window=None): v, _ = s1_screen(records) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) feats = features(v, window) m = train(v, feats) accuracy, baseline = evaluate(v, m, feats) return m, accuracy, baseline, len(v) ``` The second block defines the three gates and passes eight events through the gates in order. Seven events are on the data side: the intact run and the six corruption classes. The eighth is on the code side: bringing the feature window down from the full history to a single period. ```python # ---- GATE 1: code. Fixed mini input , never sees production data. MINI = [{"subscriber": "A001", "period": d, "region": "east", "type": "commercial", "m3": v} for d, v in (("2026-02", 30), ("2026-03", 50), ("2026-04", 10))] EXPECTED = ({"subscriber_mean": 40.0, "region_mean": 40.0, "commercial": 1}, [0, 1, 0]) def code_gate(window=None): feats = features(MINI, window) return (feats[("A001", "2026-04")], [label(k) for k in MINI]) != EXPECTED # ---- GATE 2: data. Five checks on the input of the pipeline. def data_gate(records, expected=2400, tolerance=0.05): FIELDS = ("subscriber", "period", "region", "index", "unit") keys = [(k["subscriber"], k["period"]) for k in records] findings = [] if any(any(a not in k for a in FIELDS) for k in records): findings.append("schema") if any(not isinstance(k.get("index"), int) for k in records): findings.append("type") if len(keys) != len(set(keys)): findings.append("uniqueness") if abs(len(records) - expected) / expected > tolerance: findings.append("row_count") if set((k["region"], k["period"]) for k in records) != set((b, d) for b in REGION for d in PERIODS): findings.append("partition_period") return findings # ---- GATE 3: model. The trained model is compared with the reference in release. def model_gate(m, accuracy, baseline, reference, tolerance=0.05, drop=0.0100): findings = [] if abs(m["train_rows"] - reference["train_rows"]) / reference["train_rows"] > tolerance: findings.append("train_rows") if reference["train_accuracy"] - m["train_accuracy"] > drop: findings.append("train_drop") if m["weights"] != reference["weights"]: findings.append("weight_shift") return findings, accuracy <= baseline SOURCE = generate_source() REFERENCE = run(SOURCE)[0] EVENTS = [("intact", SOURCE, None)] + [(s, corrupt(SOURCE, s), None) for s in CORRUPTIONS] \ + [("feature_code", SOURCE, 1)] RESULTS = {} print(f"{'event':<18}{'code (1)':>9}{'data (2)':>26}{'model (3)':>18}{'first gate':>11}") for name, recs, window in EVENTS: code_stop = code_gate(window) data_findings = data_gate(recs) m, accuracy, baseline, n_rows = run(recs, window) model_findings, below_baseline = model_gate(m, accuracy, baseline, REFERENCE) first = 1 if code_stop else (2 if data_findings else (3 if model_findings else 0)) RESULTS[name] = (m, accuracy, baseline, n_rows, code_stop, data_findings, model_findings, below_baseline, first) print(f"{name:<18}{('stopped' if code_stop else '-'):>9}{(','.join(data_findings) or '-'):>26}" f"{(','.join(model_findings) or '-'):>18}{(first or '-'):>11}") print("events no gate stopped:", [a for a in RESULTS if RESULTS[a][8] == 0]) print("runs whose test accuracy cannot exceed the base class ratio:", sum(1 for a in RESULTS if RESULTS[a][7]), "/", len(EVENTS)) ``` ``` event code (1) data (2) model (3) first gate intact - - - - late_arrival - - - - duplicated - uniqueness - 2 schema_changed - schema,type train_rows 2 unit_changed - - train_drop 3 missing_partition - partition_period train_rows 2 type_drift - type - 2 feature_code stopped - weight_shift 1 events no gate stopped: ['intact', 'late_arrival'] runs whose test accuracy cannot exceed the base class ratio: 8 / 8 ``` ## Which Gate Stops Which Corruption The distribution is not even. The data gate stops **four** of the six classes: the duplicated record with the uniqueness check, the field name change with the schema and type checks, the missing partition with the period-aware completeness check, the number turned into text with the type check. The model gate stops **three**, and **the unit change stops only there**: while the input of the pipeline looks flawless, the training accuracy falls from 0.7931 to **0.5919**, and the drop of **0.2012** is twenty times the tolerance of 0.0100. The code gate stops **none** of the six classes, and this is not a defect. The code gate does not see production data; the only thing it sees is a fixed input of three rows. In return, it catches bringing the feature window down to a single period **at the first step**, without touching the source at all. The model gate catches the same change too, but at the third step: it fires a weight shift because the chosen weights come out as `(0.6, 0.0, 0)` instead of `(0.6, 0.4, 0)`. Both gates are right; the difference between them is that one stops the corruption before the pipeline runs at all, and the other after 2200 rows have been processed. The last line is the most uncomfortable number of the lesson. In **eight** of the eight runs the test accuracy **cannot exceed** the base class ratio. The label distribution is imbalanced, the base class ratio is **0.7964**, and in the test set it is **0.8067**; the model produces exactly that number. A model gate that makes the test accuracy the release criterion would let no model through in this setup. This is not a false alarm: it is the evidence, invisible when accuracy is read alone, that the model never rose above the base class ratio. ## The Class No Gate Stops The table leaves two events unstopped. One is the intact run, and it is expected not to stop. The other is the **late-arriving** row, and that is a gap. The third block measures where the gap comes from. ```python print(f"{'event':<18}{'source':>8}{'pipeline':>10}{'train rows':>12}{'train acc.':>12}{'test acc.':>11}") for name, recs, window in EVENTS: m, accuracy, baseline, n_rows, *_ = RESULTS[name] print(f"{name:<18}{len(recs):>8}{n_rows:>10}{m['train_rows']:>12}" f"{m['train_accuracy']:>12.4f}{accuracy:>11.4f}") print() late_model, late_accuracy, late_baseline, *_ = RESULTS["late_arrival"] print(f"late_arrival train rows deviation: {abs(late_model['train_rows'] - 1600) / 1600:.4f}" f" gate tolerance: 0.0500 comparison: strict") print("late_arrival if the tolerance were 0.0400:", model_gate(late_model, late_accuracy, late_baseline, REFERENCE, tolerance=0.04)[0], " intact run with the same tolerance:", model_gate(RESULTS["intact"][0], RESULTS["intact"][1], RESULTS["intact"][2], REFERENCE, tolerance=0.04)[0] or "did not stop") print("runs the data gate stopped but that gave exactly the same model as the reference:", [a for a in RESULTS if a != "intact" and RESULTS[a][5] and RESULTS[a][0] == REFERENCE]) ``` ``` event source pipeline train rows train acc. test acc. intact 2400 2200 1600 0.7931 0.8067 late_arrival 2360 2120 1520 0.7914 0.8067 duplicated 2460 2200 1600 0.7931 0.8067 schema_changed 2400 1800 1200 0.7892 0.8067 unit_changed 2400 2200 1600 0.5919 0.8067 missing_partition 2354 2108 1508 0.7924 0.8067 type_drift 2400 2200 1600 0.7931 0.8067 feature_code 2400 2200 1600 0.7925 0.8067 late_arrival train rows deviation: 0.0500 gate tolerance: 0.0500 comparison: strict late_arrival if the tolerance were 0.0400: ['train_rows'] intact run with the same tolerance: did not stop runs the data gate stopped but that gave exactly the same model as the reference: ['duplicated', 'type_drift'] ``` The late-arriving row squeaks past both gates. In the data gate the source is 2360 instead of 2400; the deviation is **0.0167** and the tolerance is 0.0500. In the model gate the training rows are 1520 instead of 1600; the deviation is **exactly 0.0500** and the tolerance is also 0.0500. What lets it through is not a measurement but a **comparison sign**: because a strict inequality is used, a deviation sitting on the boundary counts as inside. When the tolerance is lowered to 0.0400, the late-arriving row stops and the intact run does not. This is the fragility every threshold-based gate carries, and it cannot be seen until the number is written. That the measured false alarm comes out zero is a property of **this set**, not of the gate: in the fictional source every period carries exactly 200 rows, so the natural variation in the row count is zero. In a source with natural variation, the same tightening would stop intact runs too. The cost of the gates is on the last line. The data gate stops four runs, but **two** of them would have produced **exactly the same** model as the reference: the duplicated record is absorbed in the deduplication stage, the number turned into text in the type conversion, and in both the weights, the training rows and the training accuracy do not change. In the setup without gates, all eight of the eight runs go to release and six are corrupted; in the setup with gates, six runs stop, two pass, and one of those that pass is still corrupted. **One third** of the runs stopped were stopped without gaining anything. The rule of the course applies to the gate too: **a number that cannot be reproduced is not a measurement but a memory.** A gate being green is not a measurement either; the late-arriving row passes all three gates green and carries the model trained with 1520 rows to release in place of the model of 1600 rows. ## Summary - The gates run in order: code **1**, data **2**, model **3**. The number of the gate that first stops an event also says how much the corruption has dirtied the pipeline. - The data gate stops **four** of the six corruption classes and the model gate **three**; the **unit change stops only at the model gate**, by dropping the training accuracy from 0.7931 to **0.5919**. - The code gate stops **none** of the six classes, because it does not see production data; in return it catches the feature code change **at the first step**, before the pipeline runs at all. - **No gate** stops the late-arriving row: the source deviation is 0.0167, the deviation in training rows is **exactly 0.0500** and the tolerance is 0.0500 with a strict comparison. With a tolerance of 0.0400 it would stop, and the intact run would not. - The label distribution is imbalanced, the base class ratio is **0.7964**; in **eight** of the eight runs the test accuracy cannot exceed the base class ratio with 0.8067. - **Two** of the four runs the data gate stopped would have produced exactly the same model as the reference; the cost of adding a gate is that it also stops where it gains nothing. ## Next Step A run that passes the gates now counts as having produced a model, but what was produced is still not written down. Up to here the model was a dictionary of three weights and a threshold, and the calling side used it by **guessing** the names, types and order of the fields. The next lesson **packages** the model: next to the weights it puts an **interface contract** that writes which fields are expected with which type and which field comes back in return. The measure is the caller that breaks when the contract is not written: how many calls drop when a field name or type changes, how many give a wrong answer without dropping at all, and how many of these the contract catches **before release**.