Continuous Integration for Machine Learning

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.

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

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

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.