33.4 Optimálás változatlan jelentéssel

33.4.1 Cél és konkrét helyzet

A VIII. ellenőrző pontban (VIII. ellenőrző pont: átadható modellprojekt) a modellprojekt átadható lett, de a predikciós pipeline fejlesztőgépen érezhetően lassú: az account_age_days kiszámítása iterrows() ciklussal történt, és 60 000 rendelésre másodperceket vesz igénybe. Ebben az alfejezetben egyetlen helyi optimalizálást végzünk el azzal a feltétellel, hogy a predikciós szerződés (Predikciós szerződés és hibateszt) semmilyen ponton nem változik: ugyanazok a feature-nevek, dtype-ok, megfigyelési egységek és toleranciakorlátok maradnak.

• Bemenet: a szintetikus Northstar-orders DataFrame order_ts és signup_date időbélyegekkel, valamint basket_minor, quantity_sum, distance_km, priority_flag oszlopokkal. A példában n=60_000 sort állítunk elő determinisztikus seeddel, mert ezen a méreten már mérhető a különbség.
• Megfigyelési egység: egy rendelés (order_id), nem ügyfél és nem tétel.
• Célkimenet: a gyorsított függvény legalább ugyanazt az account_age_days egész napokban kifejezett feature-t adja, a lebegőpontos oszlopok pedig 1e-4 relatív tűrésen belül azonosak maradnak.
• Előzetes becslés: a vektorizált változat várhatóan 20–100-szor gyorsabb; a downcast nagyjából 25–40 % DataFrame-méret-csökkenést ad. A pontos szám nem előre adott, tehát minden állítást méréssel igazolunk.

33.4.2 A kontrollháromszög: idő, memória, numerikus egyezés

Egy gyorsítás akkor nevezhető azonos jelentésű optimalizálásnak, ha három mérést egyszerre mutat fel:

1. Futásidő (wall-clock time): azonos gépen, azonos inputon, több ismétléssel. Az operációs rendszer terhelése miatt az egyetlen futás nem elegendő. A timeit.repeat által visszaadott minták közül a minimum stabilabb a zajjal szemben, mint az átlag, mert a háttérfolyamatok az átlagot felfelé torzítják, a minimum nem.
2. Memória: a DataFrame tényleges belső mérete memory_usage(deep=True).sum() és adott esetben a tracemalloc csúcsfogyasztása. A deep=True a Python-objektumok tartalmát is beszámítja, nem csak a NumPy-arrays mutatóit.
3. Numerikus egyezés (equivalence check): az eredeti aranyszabály kimenetének és a gyorsított változat kimenetének összehasonlítása meghatározott tűréssel.

A numerikus tűrés általános formulája:

$$|a_i - b_i| \le \tau_a + \tau_r \cdot |b_i|,$$

ahol $a_i$ a gyorsított, $b_i$ az aranyszabály szerinti érték; $\tau_a$ az abszolút, $\tau_r$ a relatív tűrés. A képlet mértékegysége megegyezik a mért mennyiségével, tehát distance_km esetén kilométerben adja meg az engedett eltérést. Például $\tau_a=10^{-6}$, $\tau_r=10^{-4}$ mellett 100 km-nél legfeljebb $10^{-6} + 10^{-4}\cdot 100 = 0{,}010001$ km eltérés megengedett. A NumPy np.allclose(a, b, rtol=..., atol=...) pontosan ezt a hibamodellt valósítja meg.

33.4.3 Downcast, projection, caching és vectorization

A négy optimalizálási elem közös jellemzője, hogy a megfigyelési egységet és a predikciós célváltozót egyik sem módosítja, csak a számítás módját vagy a közbülső reprezentációt.

33.4.3.1 Downcast

A pandas alapértelmezésben int64 és float64 típusokat használ, mert ezek biztonságosan lefednek minden szokásos értéket. Ha egy oszlop értékkészlete szűkebb, a dtype szűkíthető; például a Northstar quantity_sum 1 és 15 közé esik, így int16 is elegendő, a basket_minor pedig az adatszerződés szerint EUR-centben pozitív és nem haladja meg a szimuláció tartományában a int32 felső határát. A szűkebb típus kisebb memóriát foglal, gyorsabb lehet a CPU-cache miatt, és gyakran gyorsítja a NumPy/BLAS műveleteket.

A downcast azonban nem cserélhető le néma astype hívásra. Minden oszlopnál ellenőrizni kell:

import numpy as np

info = np.iinfo(np.int32)
if not (df["basket_minor"].min() >= info.min and df["basket_minor"].max() <= info.max):
    raise ValueError("basket_minor outside int32 range")

A float64 → float32 átalakítás relatív kerekítése körülbelül $2^{-24} \approx 6\cdot 10^{-8}$, tehát egyetlen értékre kicsi. Ha viszont később sok számot összegzünk, például csoportos kosárösszeget, a halmozódó hiba már túllépheti a tűrést. Pénzösszeget ezért sosem szabad float32-ben összegezni; a kosár maradjon egész minor unit formában.

33.4.3.2 Vectorization

Az eredeti lassú kód minden sorra Python-kifejezésben hajtja végre a kivonást:

def slow_age(df):
    out = []
    for _, row in df[["order_ts", "signup_date"]].iterrows():
        out.append((row["order_ts"] - row["signup_date"]).days)
    return pd.Series(out, index=df.index, dtype=np.int32)

Az iterrows() minden sorra új Series-t és Python-objektumokat hoz létre; a 60 000 sor ciklusa ezért másodpercekig fut. A vektorizált változat a pandas DatetimeIndex-ekre közvetlenül alkalmazott kivonás, majd a .dt.days accessor:

def fast_age(df):
    return (df["order_ts"] - df["signup_date"]).dt.days.astype(np.int32)

Itt a kivonás teljes tömbre vonatkozik, a .dt.days a belső C-műveletet használja, nem a Python-értelmező per-sorosan magyarázza a kódot. A mátrixszorzásos lépéseknél a NumPy a gépre telepített BLAS-t is igénybe veheti, ami további gyorsulást ad.

33.4.3.3 Projection

A projection (vetítés) azt jelenti, hogy csak azokat az oszlopokat és sorokat olvassuk be vagy tartjuk a memóriában, amelyeket a későbbi számítás ténylegesen használ. A predikciós szerződésben rögzített feature-lista alapján például nem olvassuk be a delivery_days és actual_delivery_at oszlopokat, mert azok kizárt prediktorok:

FEATURE_COLS = [
    "order_id", "order_ts", "signup_date", "distance_km",
    "priority_flag", "basket_minor", "quantity_sum",
]
orders = pd.read_csv("data/orders.csv", usecols=FEATURE_COLS)

A usecols már a CSV-értelmezés előtt szűkíti az adatmozgatást, tehát kevesebb I/O és kisebb kezdő DataFrame keletkezik. A projection a memóriában is folytatódik: az ideiglenes segédoszlopokat törölni kell, mielőtt a modellbe adnánk a táblát, mert a felesleges oszlopok is memóriát foglalnak.

33.4.3.4 Caching

Ha egy adatelőkészítő vagy feature-számító függvényt többször hívunk, a kimenet elmenthető. A cache azonban csak akkor érvényes, ha a bemenet ugyanaz. A kulcs legyen a bemeneti fájlok hash-e és a feature-verzió:

cache_key = f"features_v3_{sha256(orders_csv_bytes).hexdigest()[:16]}.joblib"

Ha a kulcs egyezik, a joblib.dump-pal elmentett fájl betölthető; ha nem, újra kell számolni. A VIII. ellenőrző pont: átadható modellprojekt alfejezetben tanult manifest- és hashfegyelem itt is kötelező: a cache-fájl nem helyettesíti a verziózott adatellenőrzést, és a splitmanifestben rögzített adatváltozatnak összhangban kell lennie a cache-kulccsal. Ellenkező esetben elavult cache-t használunk, ami a modell rosszabb teljesítményét okozhatja.

33.4.3.5 Equivalence check

Az optimalizált változatot minden releváns oszlopra összehasonlítjuk az aranyszabállyal:

• egész és kategóriás oszlopokra bitazonosság (np.array_equal),
• lebegőpontos oszlopokra np.allclose meghatározott rtol és atol értékkel,
• shape és dtype ellenőrzés a pipeline kimeneténél,
• a predikció esetén a Predikciós szerződés és hibateszt szerinti (n, 2)-es valószínűség-séma is.

A teszt nem azt kérdezi, hogy a gyorsított kód nagyjából hasonló, hanem hogy a modell tényleges bemenete és kimenete a tűrésen belül azonos-e.

33.4.4 Teljes minimális munkapélda

Az alábbi kód a szövegben szereplő Northstar-orders oszlopok mintájára előállít egy 60 000 soros szintetikus DataFrame-et, majd leméri a lassú és a gyors változatot. A mérési számok ebben a generálási módban nem valós futási jegyzék; a kód a tanuló gépén futtatandó. A kód kommentjei angolok, a változónevek a tankönyv konvencióját követik.

import timeit
import tracemalloc
import numpy as np
import pandas as pd

# Deterministic synthetic input; one row is one order.
rng = np.random.default_rng(20261020)
n = 60_000

df = pd.DataFrame({
    "order_ts": (
        pd.Timestamp("2025-01-01", tz="UTC")
        + pd.to_timedelta(rng.integers(0, 180 * 24 * 3600, size=n), unit="s")
    ),
    "signup_date": (
        pd.Timestamp("2024-06-01")
        + pd.to_timedelta(rng.integers(0, 365, size=n), unit="D")
    ),
    "basket_minor": rng.integers(1000, 50000, size=n),
    "quantity_sum": rng.integers(1, 15, size=n, dtype=np.int16),
    "distance_km": rng.gamma(2.0, 15.0, size=n),
    "priority_flag": rng.integers(0, 2, size=n, dtype=np.int8),
})

def slow_features(df):
    """Gold baseline: explicit Python loop for account_age_days."""
    out = []
    for _, row in df.iterrows():
        out.append((row["order_ts"] - row["signup_date"]).days)
    df = df.copy()
    df["account_age_days"] = pd.Series(out, index=df.index, dtype=np.int32)
    return df

def fast_features(df):
    """Optimized: vectorized datetime and checked downcasts."""
    df = df.copy()
    df["account_age_days"] = (
        (df["order_ts"] - df["signup_date"]).dt.days.astype(np.int32)
    )
    info = np.iinfo(np.int32)
    if not (df["basket_minor"].min() >= info.min
            and df["basket_minor"].max() <= info.max):
        raise ValueError("basket_minor outside int32 range")
    df["basket_minor"] = df["basket_minor"].astype(np.int32)
    df["distance_km"] = df["distance_km"].astype(np.float32)
    return df

def measure_time(func, df_arg, repeats=5, number=5):
    """Return min and median seconds per call as (min, median)."""
    raw = timeit.repeat(lambda: func(df_arg), repeat=repeats, number=number)
    per_call = [item / number for item in raw]
    return min(per_call), float(np.median(per_call))

gold = slow_features(df)
candidate = fast_features(df)

# Equivalence: integer and boolean columns require bitwise equality.
assert np.array_equal(gold["account_age_days"].to_numpy(),
                      candidate["account_age_days"].to_numpy())
assert np.array_equal(gold["basket_minor"].to_numpy(),
                      candidate["basket_minor"].to_numpy())
# Float columns use a tolerance. Distance values are synthetic km.
assert np.allclose(gold["distance_km"].to_numpy(),
                   candidate["distance_km"].to_numpy(),
                   rtol=1e-4, atol=1e-6)
print("equivalence: OK")

slow_min, slow_med = measure_time(slow_features, df)
fast_min, fast_med = measure_time(fast_features, df)
print(f"slow  min={slow_min:.4f}s median={slow_med:.4f}s")
print(f"fast  min={fast_min:.4f}s median={fast_med:.4f}s")

gold_mb = gold.memory_usage(deep=True).sum() / 1e6
cand_mb = candidate.memory_usage(deep=True).sum() / 1e6
print(f"gold DataFrame: {gold_mb:.2f} MB, candidate: {cand_mb:.2f} MB")

# Peak allocation during the optimized call.
tracemalloc.start()
_ = fast_features(df)
_, peak_bytes = tracemalloc.get_traced_memory()
tracemalloc.stop()
print(f"fast peak allocation: {peak_bytes / 1e6:.2f} MB")

Ha a függvény fut, az equivalence: OK sor akkor jelenik meg, ha mindhárom oszlopra teljesül a tűrés. A futásidő gépenként eltérő; tankönyvi szemléltetésként egy CPU-s laptopon gyakran a lassú változat 0,4–0,8 másodperc, a gyors pedig 0,003–0,010 másodperc közé esik. A DataFrame méretében a basket_minor int32-re és a distance_km float32-re alakítása nagyjából 25 % megtakarítást hoz, mert a két nagy oszlop bájtjainak egy része felszabadul.

33.4.5 Eredmény értelmezése

A mérési kódból kiderül, hogy a gyorsítás nem „feltételezett”, hanem ugyanazon az inputon, ugyanazon a gépen mért. A min-alapú futásidő csökkenti a háttérterhelés torzítását, a median a tipikus futást mutatja. Ha a kettő nagyon eltér, a mérési környezet terhelt, és érdemes a number értéket növelni.

A numerikus egyezés azért fontos, mert a lebegőpontos distance_km float32-re alakítása tényleges kerekítési hibát visz be. Az np.allclose(rtol=1e-4, atol=1e-6) ezt a hibát a tűrésen belül engedi. A tűrés megválasztása a feladat mértékegységéből következik: 1 km alatt 1 méternyi abszolút hiba sem változtatja meg a modell döntését, de 10 % relatív hiba már igen.

A quantity_sum és priority_flag oszlopok már eleve szűk dtype-tal rendelkeznek a bemenetben; a downcast itt nem hozhat további nyereséget. A gyorsított változat tehát csak azokat az oszlopokat alakítja, amelyeknél a tartományvizsgálat megengedi, és minden más értéket bitazonosan őriz.

Az optimalizálás helyességi tesztjének nemcsak az átlagot, hanem a sorazonosítókat, a hiányok szemantikáját és az időzóna-kezelést is őriznie kell.

33.4.6 Feladatok

1. Értelmezés Miért nem elegendő egyetlen wall-clock futás ahhoz, hogy kimondjuk a gyorsítást? Mi a szerepe a timeit.repeat number paraméterének, és miért a minimumot, nem az átlagot érdemes jelenteni? Fogalmazza meg két-három mondatban a mérési protokoll követelményét.

2. Megvalósítás Írjon safe_downcast függvényt, amely a basket_minor oszlopot int32-re, a distance_km-et float32-re alakítja. Ha a basket_minor értékkészlete kívül esik az int32 tartományán, a függvény dobjon ValueError-t. A distance_km-re 100 km felett is maximálisan $10^{-4}\cdot\text{érték} + 10^{-6}$ eltérést engedjen meg az equivalence checkben.

3. Hibadiagnózis Egy fejlesztő a következő gyorsítást alkalmazza: a quantity_sum oszlopot float32-re alakítja, mert így kisebb a DataFrame. A csoportos kosárösszeg egy 13456 darabos rendelési csoportban 13456.0 helyett 13455.997-ot ad, és a későbbi modell AUC-értéke 0,782-ről 0,778-ra romlik. Hol a hiba, és milyen szabályt sértett meg?

4. Megváltozott feltétel A projekt elérte az 1,2 millió rendelést; a fejlesztő a gyorsított fast_features kimenetét ügyfélazonosítónként külön particionált Parquet-fájlokba menti. A második futtatáskor csak azt a partíciót szeretné újraszámolni, ahol a bemenő vektor megváltozott. Tervezze meg a cache-kulcsot és az érvénytelenedési feltételt úgy, hogy az ügyfélazonosító a megfigyelési egység maradjon, és a numerikus equivalence minden partícióra teljesüljön. Adja meg a cache-frissítést ellenőrző pszeudokódot.

33.4.7 Megoldások és indoklás

1. Egyetlen wall-clock futás a háttérfolyamatok és az operatív memória állapota miatt nem reprodukálható. A timeit.repeat number paramétere meghatározza, hány függvényhívásból áll egy mérési ismétlés; ez csökkenti a mért időben a Python-értelmező indítási költségét, bár a loop-overhead részben megmarad. A minimum azért jobb, mint az átlag, mert a min a legkevésbé zavart futást mutatja: a process legalább ennyi idő alatt képes elvégezni a feladatot. A medián a tipikus futást jelzi. A protokoll követelménye: azonos gépet, azonos inputot, több ismétlést, azonos number-t, és a memóriaterhelést is rögzíteni kell.

2. Egy lehetséges megoldás:

def safe_downcast(df):
    df = df.copy()
    info = np.iinfo(np.int32)
    if df["basket_minor"].min() < info.min or df["basket_minor"].max() > info.max:
        raise ValueError("basket_minor does not fit int32")
    df["basket_minor"] = df["basket_minor"].astype(np.int32)
    df["distance_km"] = df["distance_km"].astype(np.float32)
    return df

# Invariant: integer column remains bit-equal, float column respects tolerance.
def check_equivalence(gold, cand):
    assert np.array_equal(gold["basket_minor"].to_numpy(),
                          cand["basket_minor"].to_numpy())
    assert np.allclose(
        gold["distance_km"].to_numpy(),
        cand["distance_km"].to_numpy(),
        rtol=1e-4,
        atol=1e-6,
    )

A meghívott np.allclose a kért képletet valósítja meg; a mértékegység kilométer. A teszt csak akkor megy át, ha minden sorban a relatív és abszolút hiba együtt a tűrés alatt van.

3. A quantity_sum darabszám egész, tehát a bitazonos reprezentáció szükséges; a lebegőpontos átalakítás beleviszi a kerekítési hibát. A hiba forrása: a float32 mantissza 24 bit, ezért nagyobb összegeknél elvesznek bináris helyiértékek. A szabály, amit megsértett: a pénz- és darabszám-egészeket sosem alakítjuk lebegőpontossá gyorsítási célból; az equivalence check csak a kategóriás és egész oszlopokra bitazonosságot fogad el, a lebegőpontos tűrés nem alkalmazható rájuk. Helyes megoldás: a quantity_sum maradjon int16, és ha a cél az AUC megőrzése, a checkpointnak minden modellbemeneti oszlopra az előírt tűrést kell vizsgálnia.

4. A partíciós cache terve:

import joblib
import hashlib

def partition_cache_key(customer_id, input_vector):
    raw = f"{customer_id}:{input_vector.tobytes()}".encode("utf-8")
    return hashlib.blake2b(raw, digest_size=16).hexdigest()

def cached_fast_features_for_customer(customer_id, df_part):
    # Load the previous key for this customer partition, if it exists.
    meta = load_partition_meta(customer_id)   # {"key": str, "file": str}
    current_key = partition_cache_key(customer_id, df_part["order_ts"].to_numpy())
    if meta is not None and meta["key"] == current_key:
        return pd.read_parquet(meta["file"])
    result = fast_features(df_part)
    new_file = f"cache/customer_{customer_id}_{current_key}.parquet"
    result.to_parquet(new_file)
    save_partition_meta(customer_id, {"key": current_key, "file": new_file})
    return result

Az érvénytelenedés feltétele: ha bármely bemeneti oszlop bytesorozata megváltozik, a blake2b kulcs megváltozik. Az ügyfélazonosító a cache-kulcs része, tehát az egyes ügyfelek megfigyelési egységét nem keverjük. A frissítés csak ahhoz az ügyfélhez törli a cache-t, amelynél a kulcsváltozást észleltük. Az equivalence-check partíciónként ugyanazokat az oszloptoleranciákat alkalmazza.

33.4.8 Gyakori tévedés és önellenőrzés

Gyorsabb, de más célt számoló program nem optimalizálás. Tipikus ellenpélda: a fejlesztő a predikciós pipeline-t úgy gyorsítja, hogy az account_age_days helyett közvetlenül a delivery_days oszlopot használja, mert az gyorsabb és jobb AUC-t ad. Ez azonban megsérti az adatszerződés azon szabályát, hogy a célmagyarázat kizárt prediktor; a gyors változat a tesztidőben nem ismert jövőbeli információt épít be. A numerikus equivalence-check önmagában nem elég: a célváltozó definícióját is változatlanul kell tartani. Hasonlóan hibás az a „gyorsítás”, amelyik a basket_minor-ból kihagyja a discount_minor levonását; a kapott kosárösszeg gyorsabban számolható, de nem ugyanaz a mennyiség.

Önellenőrző kérdések: (1) Az optimalizált változat minden modellbemeneti oszlopra azonos dtype-ot és értékkészletet tart-e? (2) A memóriamérést a deep=True vagy a shallow pointerek alapján végeztem? (3) A gyorsított függvény bemenetét ugyanazzal az adatszerződéssel validáltam-e, mint az aranyszabályt? (4) A cache-kulcs tartalmazza-e az összes felhasznált input hashét?

33.4.9 Rövid összefoglaló

Az optimalizálás csak akkor elfogadható, ha a futásidő, a memória és a numerikus kimenet egyszerre változatlan jelentést igazol. A downcast szűkíti a dtype-ot, de csak tartományellenőrzés után; a vectorization a Python-ciklus helyett tömbműveletet használ; a projection felesleges oszlopokat hagy ki; a caching elmenti a drága eredményt input-hash-hez kötve. Az equivalence-check az egész oszlopokra bitazonosságot, lebegőpontos oszlopokra relatív- és abszolút tűrést követel. A következő témakör ezt a mért, reprodukálható optimalizálási fegyelmet nagyobb adatra, a Nagyobb adat: Dask, Spark és elosztott végrehajtás Dask- és Spark-alapú környezeteire terjeszti ki.