35.1 Batch inference és idempotens output

35.1.1 Cél és konkrét helyzet

A Northstar-csapat a korábban jóváhagyott modellt, a northstar-late-v3-2025-08-01 azonosítójú champion pipeline-t, új szintetikus rendeléseken szeretné futtatni batch jelleggel. A kérdés: a rendelés pillanatában mennyi a valószínűsége annak, hogy a kézbesítés a 7 napos ígért határidőn túl érkezik. A bemenet egy CSV, amely rendelésenként csak a rendelés pillanatában ismert mezőket tartalmazza: order_id, order_ts, promised_days, quantity_sum, item_rows, basket_minor, shipping_fee_minor, currency, distance_km, priority_flag, account_age_days, segment, region és acquisition_channel. A output legyen sorazonosítóval, modellazonosítóval, modelldátummal, a késés becsült valószínűségével és a 0,5 küszöbhöz tartozó osztállyal. Önálló manifestnek kell igazolnia, hogy mely input adatverzióból, mely modellből, milyen sémaverzióval, hány sorra és hány chunkban készült a predikció. Előzetes elvárás: a kimenet sorainak száma egyezzen a bemenet rendeléseinek számával; a késés valószínűsége 0 és 1 közé essen; a késések aránya a modell tanítási osztályaránya közelében, nagyságrendileg 20% körül mozogjon. Ez számszerű cél, nem garantált futási eredmény.

35.1.2 A batch predikció mint szerződés

A batch (kötegelt) predikció nem rendelésenkénti, valós idejű szolgáltatás, hanem egyszerre több sor kiértékelése egy későbbi időpontban. A batch feladat attól még nem történelmi visszatekintés: a predikció pillanata továbbra is az order_ts. A scoring input minden sora rendelésenkénti megfigyelési egység, és minden prediktorértéknek olyannak kell lennie, amit a rendszer az adott rendelés időpontjában már ismert. Ez a feature availability követelmény.

A schema validation a Predikciós szerződés és hibateszt predikciós szerződésének batch-változata: a fontos lépés nem a modell újratanítása, hanem annak ellenőrzése, hogy az új adat ugyanazt a szerződést teljesíti. Ellenőrizendő: a feature-nevek, a dtype-ok, a hiányzó oszlopok, a hiányzó értékek és a kategorikus változók megengedett értékei. A példánkban a modell a következő gyakorlati X mátrixot várja: $n$ rendelés és $p=9$ oszlop, ahol a numerikus mezők distance_km (km, float64, ≥ 0), priority_flag (0 vagy 1), basket_minor (EUR-cent, int64), quantity_sum (darab, pozitív egész), account_age_days (nap, int64), a kategorikus mezők pedig segment, region és acquisition_channel. A modell predict_proba kimenete mindig $(n, 2)$ alakú mátrix; az oszlopok $[\hat p_0, \hat p_1]$ soronként 1-re összegződnek, ahol $\hat p_1$ a késés becsült valószínűsége. A döntési osztály: 1, ha $\hat p_1 \ge 0{,}5$, különben 0.

A chunk a Nagyobb mint RAM: előbb adatfolyam adatfolyam-fogalmának alkalmazása itt: a nagy CSV-t nem egyszerre kell a memóriában tartani, hanem a pandas chunksize paraméterével darabokban. Minden chunknak át kell esnie ugyanazon a séma-ellenőrzésen, mert a chunkolás önmagában nem őrzi meg a dtype-ot egy hibás CSV-ben; a második chunkban megváltozó tartalom eltérő dtype-ot eredményezhet. A chunkok sorrendjét és a predikciós számítást megőrizzük.

Az output manifest a Futásazonosító és kísérleti táblázat kísérleti futásmanifest fogalmának batch-szállítási változata. Az identitás most a bemeneti adatverzióból, a modellartifact hashekből és a sémaverzióból áll, ezért tartozik az output manifest az output mellé. Az output manifest nem tartalmazza a jövőbeli címkéket, azaz a delivery_days, actual_delivery_at, label_available_at és target_late mezőket meg sem próbáljuk becsülni. A manifest sem hitelesíti a modell minőségét: a modellkártya határai, például a régiók támogatott köre, a batch futástól függetlenül érvényesek.

35.1.3 Teljes minimális munkapélda

A példa egy helyi, CPU-s CLI, amely a lab projektben fut. A mentett pipeline-t joblib betölti; ez a saját projectből származó artefaktum, nem ismeretlen pickle. A kód determinisztikus, és a CSV-kimenet hash-e a manifestben megismételhetővé válik.

Először a scoring input CSV, amely az orders.csv szerződéséből csak a rendeléskor ismert mezőket tartalmazza. Hat sor elegendő a chunkolás bemutatásához:

# artifacts/new_orders_scoring_input.csv (szemléltető részlet)
# order_id,order_ts,promised_days,quantity_sum,item_rows,
# basket_minor,shipping_fee_minor,currency,distance_km,
# priority_flag,account_age_days,segment,region,acquisition_channel
# ord-9001,2025-05-02T08:12:00+00:00,7,2,1,2490,490,EUR,14.6,1,123,consumer,north,search
# ord-9002,2025-05-02T09:03:00+00:00,7,1,1,990,290,EUR,42.1,0,88,business,south,direct
# ord-9003,2025-05-03T13:40:00+00:00,7,3,2,3110,590,EUR,8.7,1,200,consumer,central,referral
# ord-9004,2025-05-03T15:22:00+00:00,7,4,1,5410,990,EUR,27.4,1,17,consumer,north,direct
# ord-9005,2025-05-04T10:01:00+00:00,7,2,2,1775,390,EUR,91.0,0,304,business,central,search
# ord-9006,2025-05-04T11:47:00+00:00,7,1,1,1290,290,EUR,63.8,0,151,consumer,south,referral

A betöltött modell a model_candidate_v3.joblib fájl. A modelldátum a jóváhagyott modell nyilvántartásából származó érték, nem a batch futás napja. A modellt a Modelljelölt, jóváhagyás és registry szerint a registry hash-ével kell ellenőrizni; a minimális példában a manifest az artifact aktuális hash-ét rögzíti. A CLI:

# lab/batch_predict.py
import argparse
import hashlib
import json
import sys
from datetime import datetime, timezone
from pathlib import Path

import joblib
import pandas as pd

FEATURE_COLUMNS = [
    "distance_km", "priority_flag", "basket_minor",
    "quantity_sum", "account_age_days",
    "segment", "region", "acquisition_channel",
]

FORBIDDEN_COLUMNS = {
    "delivery_days", "actual_delivery_at",
    "label_available_at", "target_late",
}

CATEGORY_RULES = {
    "segment": {"consumer", "business"},
    "region": {"north", "central", "south"},
    "acquisition_channel": {"search", "referral", "direct"},
}

DTYPE_RULES = {
    "distance_km": "float64",
    "priority_flag": "int64",
    "basket_minor": "int64",
    "quantity_sum": "int64",
    "account_age_days": "int64",
    "segment": "object",
    "region": "object",
    "acquisition_channel": "object",
}


def sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for block in iter(lambda: f.read(8192), b""):
            h.update(block)
    return h.hexdigest()


def validate_schema(df: pd.DataFrame) -> list[str]:
    errors = []
    missing = [c for c in FEATURE_COLUMNS if c not in df.columns]
    if missing:
        errors.append(f"missing feature columns: {missing}")
        return errors
    forbidden = [c for c in FORBIDDEN_COLUMNS if c in df.columns]
    if forbidden:
        errors.append(
            f"forbidden future-only columns present: {forbidden}"
        )
    for col, expected_dtype in DTYPE_RULES.items():
        if df[col].dtype != expected_dtype:
            errors.append(
                f"{col}: expected dtype {expected_dtype}, got {df[col].dtype}"
            )
    for col, allowed in CATEGORY_RULES.items():
        unknown = set(df[col].astype("string").dropna()) - allowed
        if unknown:
            errors.append(f"{col}: unknown categories {sorted(unknown)}")
    if df[FEATURE_COLUMNS].isna().any().any():
        errors.append("missing values in feature columns")
    return errors


def main(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", type=Path, required=True)
    parser.add_argument("--model", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--chunk-size", type=int, default=10_000)
    parser.add_argument("--model-id", default="northstar-late-v3-2025-08-01")
    parser.add_argument("--schema-version", default="northstar-v3")
    args = parser.parse_args(argv)

    model = joblib.load(args.model)
    model_datum = "2025-08-01"  # a regisztrált modell jóváhagyási dátuma

    input_hash = sha256_file(args.input)
    model_hash = sha256_file(args.model)
    total_rows = 0
    chunk_count = 0

    reader = pd.read_csv(args.input, chunksize=args.chunk_size)
    for chunk in reader:
        chunk_count += 1
        errors = validate_schema(chunk)
        if errors:
            raise SystemExit(
                f"chunk {chunk_count} failed schema validation: {errors}"
            )
        X = chunk[FEATURE_COLUMNS]
        proba = model.predict_proba(X)[:, 1]
        out_chunk = pd.DataFrame({
            "order_id": chunk["order_id"],
            "model_id": args.model_id,
            "model_datum": model_datum,
            "predicted_probability_late": proba,
            "predicted_class": (proba >= 0.5).astype("int64"),
        })
        out_chunk.to_csv(
            args.output,
            mode="w" if chunk_count == 1 else "a",
            header=(chunk_count == 1),
            index=False,
        )
        total_rows += len(chunk)

    output_hash = sha256_file(args.output)
    output_rows = pd.read_csv(args.output).shape[0]
    manifest = {
        "input_file": str(args.input),
        "input_sha256": input_hash,
        "model_file": str(args.model),
        "model_sha256": model_hash,
        "model_id": args.model_id,
        "model_datum": model_datum,
        "schema_version": args.schema_version,
        "chunk_count": chunk_count,
        "input_record_count": total_rows,
        "output_record_count": output_rows,
        "output_sha256": output_hash,
        "created_at_utc": datetime.now(timezone.utc).isoformat(),
        "python": sys.version.split()[0],
    }
    manifest_path = args.output.with_suffix(".manifest.json")
    manifest_path.write_text(
        json.dumps(manifest, indent=2, ensure_ascii=False) + "\n"
    )
    print(
        f"rows={total_rows} chunks={chunk_count} "
        f"output={args.output} manifest={manifest_path}"
    )


if __name__ == "__main__":
    main()

A futtatási parancs Linux/macOS terminálból, a lab munkakönyvtárból:

cd lab
python batch_predict.py \
  --input artifacts/new_orders_scoring_input.csv \
  --model artifacts/model_candidate_v3.joblib \
  --output artifacts/new_orders_scoring_output.csv \
  --chunk-size 2

A --chunk-size 2 csak demonstráció; éles futásban a méretet memóriaprofilonként érdemes megválasztani. A szemléltető, automatikusan még nem ellenőrzött kimenet: rows=6 chunks=3 output=artifacts/new_orders_scoring_output.csv manifest=artifacts/new_orders_scoring_output.manifest.json.

A kimeneti CSV a következő oszlopokkal rendelkezik, a sorok a bemenet sorrendjében jelennek meg:

A manifest legfontosabb mezői a következő ellenőrzési láncot adják. Az input_sha256 az adatverzió; a model_sha256 az artifact; a model_id és model_datum a regisztrált modell; az output_sha256 a kimenet tartalmi ellenőrzésére szolgál. A created_at_utc nem az output identitásának része, hanem a futás auditrekordja. A manifestet táblázatban összefoglalva:

35.1.4 Eredmény értelmezése

A kimenet nem tartalmazhat valós jövőbeli címkét, ezért a pontosság itt nem mérhető. Az ellenőrzés ehelyett szerződést ellenőriz: a sorok száma, a valószínűségek tartománya, a dtype-ok és a manifest mezői. A minimális példában az invariánsok:

• output_record_count == input_record_count == 6;
• 0 <= predicted_probability_late <= 1 minden sorban;
• predicted_class pontosan akkor 1, ha a valószínűség legalább 0,5;
• a kimenet oszlopai nem tartalmaznak tiltott future-only mezőt;
• a modell és a séma verziója megegyezik a futtatási paranccsal.

Az idempotens output azt jelenti, hogy ugyanazzal a bemenettel, modellel, modelldátummal, sémaverzióval és chunkonkénti sorrenddel a kimeneti CSV tartalma azonos, tehát a hash-e azonos. A created_at_utc kivétel, mert az az audit időpontja. Ha a folyamat a második chunk közepén megszakad, az output félkész; a helyreállítás az egész parancs újrafuttatása, mert a kód az első chunknál felülírja a fájlt. A termelési szintű CLI ezt egy ideiglenes fájlba írással és atomi átnevezéssel teszi robusztusabbá, mint a VIII. ellenőrző pont: átadható modellprojekt recovery-elvének batch-változata.

Fontos érvényességi korlát: a manifest a futás azonosságát bizonyítja, de nem bizonyítja, hogy a modell továbbra is jól működik. Ha az új rendelések eloszlása eltér a tanítási ablaktól, a hash és a séma-kontroll önmagában nem jelzi; ahhoz az Adat- és predikciómonitoring címke nélkül monitorozási folyamata szükséges. A Northstar-példa szintetikus, ezért a táblázat számai szemléltetőek, nem mért eredmények.

Az idempotens batch outputhoz stabil bemeneti kulcs, a futásverzió ellenőrzése és atomikus publikálás szükséges; az azonos fájlnév önmagában nem elegendő.

35.1.5 Feladatok

1. Értelmezés. A batch manifest szerint input_sha256=9f8c..., model_sha256=a441..., schema_version=northstar-v3, output_record_count=2400. Melyik állítás igaz? a) Ezzel a manifesttel bizonyítottuk, hogy a modell a 2400 rendelés 20%-ában késést jósolt. b) A manifest csak azt bizonyítja, hogy a kimenet ebből a bemenetből, ezzel a modellel, ezzel a sémaverzióval készült, és a kimenet sortartalma a hash-sel újra ellenőrizhető. c) Az output_record_count mindig az input sorainak kétszerese, mert a modell pozitív és negatív osztályra is számol. Indokold a választ.

2. Megvalósítás. Írj egy check_batch_contract(input_manifest: dict, output_path: Path) -> None függvényt, amely a manifestből kiolvassa az input_sha256 és output_sha256 értékeket, majd ellenőrzi, hogy az input- és outputfájl aktuális hash-e egyezik a manifesttel. Eltérés esetén ValueError-t dobjon. A fájl elérési útja a manifest input_file és a kimeneti CSV adott kulcsából jön. Add meg a hiányzó fájl kezelését is.

3. Hibakeresés. Egy batch futás az első chunknál hibátlan, a másodiknál a következő üzenetet adja: ValueError: X has 9 features, but LogisticRegression is expecting 10 features as input. A fejlesztő megnézi a CSV-t, és azt mondja, hogy a fájl elején minden rendben van. Mi lehet a hiba? Hogyan helyes a parancssor diagnosztikája? Mit kell megváltoztatni a hívó kódban, hogy a hiba már a chunk betöltésekor egyértelmű legyen?

4. Megváltozott feltétel. Az input CSV-ben egy új régióérték jelenik meg: west. A validate_schema az ismeretlen kategóriát hibaként jelzi. Hogyan kezeled? Indokold, miért nem elég a modell handle_unknown="ignore" beállítása a további futtatáshoz, és milyen modellnyilvántartási lépések következnek.

35.1.6 Megoldások és indoklás

1. A helyes válasz b). Az output_record_count nem metrika; a 2400 sor kimeneti sort jelez, nem a késések számát. A manifest a láncolat azonosságát igazolja, nem a modell pontosságát. A c) hamis, mert a predikció kimenete soronként egy valószínűség, nem két sor. Az a) hamis, mert címke nélkül nem tudjuk, hány sor késik; a manifest akkor sem mondja meg, ha a modell osztályt is ad.

2. Lehetséges megoldás:

import hashlib
from pathlib import Path


def sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for block in iter(lambda: f.read(8192), b""):
            h.update(block)
    return h.hexdigest()


def check_batch_contract(manifest: dict, output_path: Path) -> None:
    input_path = Path(manifest["input_file"])
    output_path = Path(output_path)
    if not input_path.exists():
        raise FileNotFoundError(f"input file missing: {input_path}")
    if not output_path.exists():
        raise FileNotFoundError(f"output file missing: {output_path}")
    if not Path(manifest["model_file"]).exists():
        raise FileNotFoundError(f"model file missing: {manifest['model_file']}")
    if sha256_file(input_path) != manifest["input_sha256"]:
        raise ValueError("input_sha256 mismatch")
    if sha256_file(output_path) != manifest["output_sha256"]:
        raise ValueError("output_sha256 mismatch")

Az ellenőrzés akkor hasznos, ha a manifestet a futás végén ugyanaz a környezet hozta létre; az útvonalak gépek között eltérhetnek, ilyenkor a relatív projektgyökérre kell normalizálni.

3. A ValueError: X has 9 features... üzenet a scikit-learn modell elvárásából jön, nem a CSV-sor számából. A fejlesztő a fájl elejét látja jónak, de a második chunkban a pandas a CSV egyik oszlopát eltérő néven olvassa be, vagy egy oszlopnév eltolódik, például egy sorban a region oszlop neve region; lett. A FEATURE_COLUMNS kiválasztás így nem az elvárt 10 oszlopot adja, a validátor missing feature columns hibát generál, de ha a hívó nem szakítja meg a futást, a modell betanított 10 feature-elvárása 9 bemenő oszlopnál a fenti hibát adja. A diagnózis: a chunk columns.tolist() kiírása, a CSV középső sorainak megnézése, adott esetben a read_csv-nek a fejléc sorát rögzítő paraméter. A javítás a hívó oldalon: a validate_schema hiba esetén raise SystemExit azonnal, ne csak naplózzon. A chunkonkénti séma-ellenőrzés pontosan ezt a hibát fogja meg emberi félreértés nélkül.

4. A west régió esetén a helyes döntés a batch futtatás felfüggesztése, nem a kategória csendes figyelmen kívül hagyása. A handle_unknown="ignore" a Predikciós szerződés és hibateszt tanítása szerint nullákkal kódolja az ismeretlen értéket, tehát a modell képes predikciót adni; a modellkártya azonban a north, central, south értékeket támogatja. A west megjelenése szerződésváltozás: előbb ellenőrizni kell, hogy az új régió a valós adatfelvételben létezik-e, majd a modell újratanítása és új modellkártya, új modelldátum szükséges. A batch kimeneten nem szabad úgy túllépni, hogy az alacsony bizalomjelzés nélkül jelenik meg. Ez a Robusztusság és korlátokkal írt modellkártya alkalmazott robusztussági elve: a modellkártyán kívül eső tartományban a futtatást meg kell állítani, nem automatikusan kiterjeszteni.

35.1.7 Gyakori tévedés ellenpéldával

Gyakori tévedés: az új scoring CSV-ben a célmező is szerepel, például a target_late oszlop, mert az elemző a historikus orders.csv-ból nyert tesztet adta át, és a modell csak nem válogatja ki. A dimensionality ellenőrzés ezt nem feltétlenül jelzi, ha a modell a feature-listából dolgozik; a batch output ugyanazzal a shape-vel elkészül, ezért a hiba láthatatlan marad. Az ellenpélda: egy 6 soros input CSV target_late oszloppal a validate_schema tiltott-oszlop ellenőrzése nélkül hibátlanul futna; az output manifest hash-e is egyezne, de a folyamat megsértette a predikció pillanatának szerződését. Ha ezt a folyamatot a későbbi éles scoring előtt a rendszer nem vizsgálja, a modell olyan adaton tanulhat, amely a rendelés pillanatában nem állt rendelkezésre; a batch CLI-nek ezért a FORBIDDEN_COLUMNS halmaz explicit, nem a feature-válogatásra bízott. A célmező jelenléte általában arra utal, hogy rossz tárterületről érkezett az input.

35.1.8 Önellenőrzés

1. Milyen adat lehet a prediktor része a Northstar batch scoringjában? Milyen oszlopok tiltottak, és miért?
2. Ha az output CSV hash-e azonos két futásban, de a manifest created_at_utc értéke eltér, idempotens-e a batch folyamat? Indokold.
3. Mi a különbség a manifest input_sha256 és output_sha256 szerepe között?

A rövid ellenőrző válaszok: 1. Rendeléskor ismert mezők; a delivery_days és társai azért tiltottak, mert a predikció pillanatában nem ismertek. 2. Igen, tartalmi értelemben idempotens; a manifest időbélyeg audit adat, nem predikciós tartalom. 3. Az input hash az adatverziót, az output hash a kimenet tartalmi egyezését ellenőrzi.

35.1.9 Rövid összefoglaló

A batch inference akkor megbízható, ha a bemenetet adatverzióhoz, a modellt modelldátumhoz kötjük, és a kimenetet manifesttel zárjuk. A sémavalidáció soronként ellenőrzi a feature-neveket, dtype-okat és kategóriákat; a tiltott jövőbeli céloszlopok jelenléte azonnali hiba. A chunkolás a memóriát korlátozza, de a séma-ellenőrzést minden chunkra futtatni kell. Az idempotens kimenet azonos bemenet, modell és séma mellett azonos CSV-hash-t ad; a manifest időbélyeg csak audit. A következő alkalmazás a FastAPI végpont és ellenőrzött bemenet FastAPI-végpont, ahol ugyanez a szerződés rendelésenkénti, kis késleltetésű szolgáltatássá válik.