30.1 Feldolgozási gráf és idempotencia

A Northstar kezdőprojektben a feladat rendeléskor becsülni, hogy a kézbesítés meghaladja-e a 7 napot. Az adatfolyam eddig egy notebook celláiban élt: beolvasás, tisztítás, feature-előállítás, modell. Ismételt futtatáskor azonban nem mindig tudtuk, hogy a orders_raw.csv-ből készült tisztított tábla hova került, és egy újraindítás néha duplikált sorokat okozott. A cél most az, hogy a feldolgozást három, egymásra épülő lépésre bontsuk, amelyek mindegyike explicit bemeneti és kimeneti fájlhatárral rendelkezik, és a teljes folyamat kétszeri futtatása nem változtat az eredményen.

Az input a mellékelt lab/raw/orders_raw.csv, lab/raw/customers.csv és lab/raw/order_items.csv; a megfigyelési egység a rendelés (order_id). A cél egy rendelésszintű feature-tábla előállítása. Előzetes becslés: a tisztított rendeléstábla a kontroll orders.csv-vel egyezik (2400 sor), a feature-táblának ugyancsak 2400 sora lesz, és minden lépés kimenete determinisztikus.

30.1.2 Feldolgozási gráf: lépések, rétegek és függőségek

A pipeline-t irányított aciklikus gráfként (DAG, directed acyclic graph) modellezzük. A csomópontok a lépések, az élek a függőségek: egy lépés csak akkor futhat, ha minden bemenete már létezik. Aciklikus, mert nincs körkörös hivatkozás; a gráf topologikus sorrendben futtatható.

A nyilak egyben a lépések be- és kimeneti határait jelölik. Egy lépés dependency-je az a fájl vagy tábla, amelyet olvas; a kimenete az a fájl, amelyet atomikusan kiír. Ha a bemenet változik, a lépést újra kell futtatni; ha nem, a kimenet változatlan marad.

30.1.2.1 Függvénykompozíció és idempotencia

f_1: \text{Raw} \to \text{Validated}, \quad f_2: \text{Validated} \times \text{Customers} \times \text{Items} \to \text{Features},

és a teljes folyamat a kompozíció:

f_2(f_1(x), y, z)

. A lépések tiszta függvényként viselkednek: ugyanaz a bemenet ugyanazt a kimenetet adja — ezt a Modulhatár és tiszta függvény alfejezetben már tárgyaltuk.

Az idempotencia azt jelenti, hogy egy lépés többszöri alkalmazása nem változtat az eredményen:

T(T(x)) = T(x)

. A mi esetünkben ez úgy teljesül, hogy a lépés a kimeneti fájlt felülírja, nem hozzáfűzi. Ha a lépés INSERT-et vagy append-et használna, akkor

T(x)

után

T(T(x))

kétszer annyi sort tartalmazna — ez az „additív adatrombolás” ellenpéldája.

30.1.2.2 Relációs szemcsézettség

Minden lépésnél rögzíteni kell a sorok jelentését. A raw és a validated rétegben egy sor egy rendelés; a customers.csv-ben egy sor egy ügyfél; az order_items.csv-ben egy sor egy rendelési tétel. A features réteg is rendelésszintű, ezért a tételszintű táblát a join előtt aggregálni kell. Ez az II. ellenőrző pont: adatból ellenőrzött feature-tábla alfejezetben megismert szabály: ha a tételsorokat közvetlenül joinolnánk a rendelésekhez, a rendelésszintű összegek (pl. shipping_fee_minor) többszöröződnének, és a szemcsézettség megváltozna. A lépés kontrollja ezért a sorok száma, az egyedi kulcsok száma és a kontrollösszegek.

30.1.2.3 Atomic output

A fájl kiírása atomi, ha a lépés először egy ideiglenes fájlba ír, majd os.replace segítségével cseréli a végleges fájlra. Azonos fájlrendszeren a névcsere tipikusan atomikus, így párhuzamos olvasó a régi vagy az új fájlnevet látja. Ez nem azonos a tartós lemezre írás garanciájával (fsync), és hálózati fájlrendszeren vagy platformok között további megszorítások lehetnek.

30.1.3 Teljes minimális példa: három lépés Pythonban

A következő pipeline.py a lab/ munkakönyvtárból futtatható Linux, macOS vagy WSL alatt. A kód pandas-t és PyArrow-t használ; a kimeneti Parquet-fájlok a lab/artifacts/ alatti rétegmappákba kerülnek. A kód nem igényel hálózatot, és a Northstar kezdőadatokkal dolgozik.

# lab/pipeline.py
from pathlib import Path
import hashlib
import json
import os
import tempfile

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

RAW = Path("raw")
ART = Path("artifacts")
VALIDATED = ART / "validated"
FEATURES = ART / "features"

def atomic_write_parquet(df: pd.DataFrame, path: Path) -> None:
    """Atomikusan ír egy DataFrame-et Parquet-fájlba."""
    path.parent.mkdir(parents=True, exist_ok=True)
    table = pa.Table.from_pandas(df, preserve_index=False)
    with tempfile.NamedTemporaryFile(
        dir=path.parent, suffix=".parquet.tmp", delete=False
    ) as tmp:
        pq.write_table(table, tmp.name)
        os.replace(tmp.name, path)

def step_raw() -> None:
    """1. lépés: nyers CSV-k betöltése, séma-ellenőrzés, raw réteg kiírása."""
    orders_raw = pd.read_csv(
        RAW / "orders_raw.csv",
        parse_dates=["order_ts"],
        dtype={"customer_id": "string", "currency": "string"},
    )
    customers = pd.read_csv(RAW / "customers.csv", dtype={"customer_id": "string"})
    items = pd.read_csv(RAW / "order_items.csv", dtype={"currency": "string"})
    # Sémaellenőrzés: a kötelező oszlopok megléte.
    required = ["order_id", "customer_id", "order_ts", "basket_minor"]
    missing = [c for c in required if c not in orders_raw.columns]
    if missing:
        raise ValueError(f"Hiányzó oszlopok az orders_raw.csv-ben: {missing}")
    # A raw rétegbe Parquet-ként írjuk, explicit sémával.
    atomic_write_parquet(orders_raw, ART / "raw" / "orders_raw.parquet")
    atomic_write_parquet(customers, ART / "raw" / "customers.parquet")
    atomic_write_parquet(items, ART / "raw" / "order_items.parquet")
    # Hash-manifest a reprodukálhatósághoz.
    manifest = {
        "orders_raw_hash": hashlib.sha256(
            (RAW / "orders_raw.csv").read_bytes()
        ).hexdigest(),
        "customers_hash": hashlib.sha256(
            (RAW / "customers.csv").read_bytes()
        ).hexdigest(),
        "order_items_hash": hashlib.sha256(
            (RAW / "order_items.csv").read_bytes()
        ).hexdigest(),
    }
    (ART / "raw" / "manifest.json").write_text(json.dumps(manifest, indent=2))

def step_validate() -> None:
    """2. lépés: tisztítás és érvényességi szűrések a raw rétegből."""
    orders = pd.read_parquet(ART / "raw" / "orders_raw.parquet")
    customers = pd.read_parquet(ART / "raw" / "customers.parquet")
    # Érvénytelen sorok azonosítása a Northstar szerződés szerint:
    # negatív kosár, hiányzó distance_km, ismeretlen customer_id.
    known_customers = set(customers["customer_id"])
    invalid_mask = (
        (orders["basket_minor"] < 0)
        | orders["distance_km"].isna()
        | ~orders["customer_id"].isin(known_customers)
    )
    invalid_rows = orders.loc[invalid_mask].copy()
    # Extra duplikált sorok eltávolítása: az order_id egyedi marad.
    deduped = orders.loc[~invalid_mask].drop_duplicates(subset=["order_id"])
    # Auditnapló a döntésekről.
    audit = {
        "input_rows": len(orders),
        "invalid_rows_removed": len(invalid_rows),
        "duplicate_rows_removed": len(orders) - len(invalid_rows) - len(deduped),
        "output_rows": len(deduped),
        "invalid_order_ids": invalid_rows["order_id"].tolist(),
    }
    (VALIDATED / "audit.json").write_text(json.dumps(audit, indent=2))
    atomic_write_parquet(deduped, VALIDATED / "orders.parquet")

def step_features() -> None:
    """3. lépés: rendelésszintű feature-tábla, az aggregáció a join előtt."""
    orders = pd.read_parquet(VALIDATED / "orders.parquet")
    customers = pd.read_parquet(ART / "raw" / "customers.parquet")
    items = pd.read_parquet(ART / "raw" / "order_items.parquet")

    # Tételszintű aggregáció rendelésszintre, mielőtt joinolnánk.
    item_agg = (
        items.groupby("order_id", as_index=False)
        .agg(
            line_total_minor_sum=("line_total_minor", "sum"),
            quantity_sum_check=("quantity", "sum"),
        )
    )

    # Joinok: csak azok a rendelések maradnak, amelyekhez van ügyfél.
    features = (
        orders.merge(customers, on="customer_id", how="inner")
        .merge(item_agg, on="order_id", how="left")
    )

    # account_age_days az order_ts és signup_date különbsége.
    features["account_age_days"] = (
        features["order_ts"] - pd.to_datetime(features["signup_date"])
    ).dt.days

    # A starter feature-oszlopok kiválasztása; a célváltozó később csatolódik.
    feature_cols = [
        "order_id", "order_ts", "promised_days", "quantity_sum", "item_rows",
        "basket_minor", "shipping_fee_minor", "distance_km", "priority_flag",
        "account_age_days", "segment",
    ]
    features = features.loc[:, feature_cols]
    atomic_write_parquet(features, FEATURES / "features.parquet")

def run_pipeline() -> None:
    """A három lépés topologikus sorrendben."""
    step_raw()
    step_validate()
    step_features()

if __name__ == "__main__":
    run_pipeline()

cd lab
python pipeline.py

python pipeline.py
python - <<'PY'
import pandas as pd
from pathlib import Path
import hashlib
p = Path("artifacts/features/features.parquet")
df = pd.read_parquet(p)
print("sorok:", len(df))
print("egyedi order_id:", df["order_id"].nunique())
print("tartalom hash:", hashlib.sha256(pd.util.hash_pandas_object(df, index=True).values.tobytes()).hexdigest()[:16])
PY

A fenti kimenet szemléltető, a könyvírási folyamat nem futtatta ezt a kódot. A laborban a várt értékek: a validated réteg a kontroll orders.csv-vel egyezik meg; a features sorainak száma 2400, és minden order_id pontosan egyszer szerepel. A másodszori futtatás után ugyanaz a sor- és kulcsszám, valamint ugyanaz a tartalom-hash adódik, mert a lépések felülírnak és nem fűznek hozzá.

30.1.4 Eredmény értelmezése és kézi kontroll

A relációs szemcsézettség itt végig rendelésszintű; a tételszintű aggregáció a step_features elején történik, így a shipping_fee_minor nem duplázódik. A hibás sorokról szóló döntéseket az audit.json rögzíti, így a tisztítás nem csendes.

30.1.5 Feladatok

30.1.6 Megoldások és indoklás

def check_idempotency() -> None:
    df = pd.read_parquet("artifacts/features/features.parquet")
    assert len(df) == 2400, f"a features sorainak száma {len(df)}, nem 2400"
    assert df["order_id"].is_unique, "az order_id nem egyedi"
    expected_basket_sum = 123456789  # a kontrollfájlból számolt érték
    assert df["basket_minor"].sum() == expected_basket_sum, "a basket_minor összege eltér"

Az expected_basket_sum értékét a laborban a tiszta orders.csv alapján kell kiszámolni; a példa itt csak a szerkezetet mutatja. Az ellenőrzés akkor megalapozott, ha a kontrollérték előre rögzített, és a futás után összehasonlítjuk.

30.1.7 Gyakori tévedés és ellenpélda

A leggyakoribb hiba, hogy egy lépés kimenetét a saját bemenetéhez fűzik, vagy a rétegeket összekeverik. Például egy jegyzetfüzetben valaki a validated táblát a step_validate cellájában így menti: orders.to_csv("validated/orders.csv", mode="a"). Az első futtatás rendben van; a második futtatás azonban duplikált sorokat hoz létre, és a feature-összegek kétszeresére nőnek. Ez a „véletlenül additív adatrombolás”: az eredmény nem a tisztított adat, hanem annak kétszerese. A helyes minta az atomic overwrite és a rétegek szigorú elkülönítése. Ez visszavezet a Modulhatár és tiszta függvény tiszta függvény és I/O határ elvéhez: a transzformáció tiszta, a mellékhatás csak a kimeneti fájl atomi cseréje.

Önellenőrzés: az egyes lépések bemeneti és kimeneti szemcsézettsége explicite azonos? A kimeneti fájl mindig ugyanazon az útvonalon jön létre? A tételszintű adat a join előtt aggregálódik? A drop_duplicates szűrés sorrendje nem rejti el a hibás sorokat? Ha ezekre igen a válasz, a pipeline idempotens.

30.1.8 Rövid összefoglaló

A feldolgozási gráf lépései explicit be- és kimeneti fájlhatárral rendelkeznek, és a raw/validated/features rétegek egymásra épülnek. A lépések tiszta függvényként viselkednek, a kimenetet atomikusan felülírják, ezért kétszeri futtatás nem duplikál sorokat. Az idempotenciát sor-, kulcs- és kontrollösszeg-invariantek ellenőrzik. A relációs szemcsézettség eközben végig rendelésszintű marad, mert a tételszintű aggregáció a join előtt történik. A dbt és az orchestratorok ezt a DAG-fegyelmet skálázzák, de kis helyi pipeline-hoz egy Python script is elég. A következő alkalmazás az Adatminőségi kapu és hibás sorok kezelése adatminőségi kapu: itt azt vizsgáljuk, hogyan állítható meg a pipeline automatikusan, ha a tisztítási feltételek sérülnek.

30.1.1 Helyzet: a kezdőprojekt egyetlen notebookból három újraindítható lépésbe