30.3 Cache, backfill és orchestrator

30.3.1 Cél és induló helyzet

Az Adatminőségi kapu és hibás sorok kezelése alfejezetben felépített adatminőségi kapu megállítja a hibás sorokat. Most a folyamat többi részét tesszük újrafuttathatóvá: ha ugyanazt a feladatot ugyanazzal a bemenettel és paraméterrel futtatjuk, ne számoljunk újra, de ha bármelyik megváltozik, a régi gyorsítótár-kimenet (cache output) ne használódjon fel. A gyakori tévedés pont ez: attól, hogy egy output fájl létezik, még nem biztos, hogy az aktuális bemenethez készült.

A fejezet laborja egy kis, helyi DAG három feladattal:

  1. extract: CSV betöltése,
  2. filter: sorok szűrése egy paraméter szerint,
  3. aggregate: ügyfelenkénti mennyiségösszeg.

Bemenetünk a következő hatrendeléses miniállomány, amelyet a kód maga hoz létre:

order_id customer_id quantity_sum priority_flag
o1 c1 3 1
o2 c1 2 0
o3 c2 5 1
o4 c2 1 1
o5 c1 4 0
o6 c3 2 0

Az order_ts oszlop is szerepel a CSV-ben, de a szűréshez most nem kell. A futási paraméter only_priority: ha 1, csak a priority_flag=1 sorokat tartjuk meg; ha 0, a többit.

Induló becslés. Mielőtt a kódot megnéznéd, számold ki kézzel: only_priority=1 esetén mely sorok maradnak, és mennyi az ügyfelenkénti quantity_sum összeg? Ugyanezt tedd meg only_priority=0-ra. Ez lesz a cache-megvalósítás kézi kontrollja.

30.3.2 Cache key, input hash és érvénytelenítés

A gyorsítótárban egy feladat eredményét egy kulcs (cache key) azonosítja. A kulcsnak legalább három dolgot kell tartalmaznia:

A tartalomhash lényege: ugyanaz a fájlnév idővel más tartalmat jelölhet, ezért a fájlnév nem elég. A kulcsot célszerű egy determinisztikus kriptográfiai hashbe tömöríteni:

k=SHA256(canonical(T,v,θ,h(x1),…,h(xn))),k = \mathrm{SHA256}\bigl(\text{canonical}(T, v, \theta, h(x_1),\dots,h(x_n))\bigr),

ahol TT a feladat neve (string), vv a feladat verziója (string), θ\theta a paraméterek kanonikus JSON alakja, h(xi)h(x_i) pedig a bemeneti fájl SHA-256 hashének hexadecimális alakja. Az SHA-256 kimenete 64 hexa karakter; a gyakorlatban csak ennyit látunk, nem a bemenetet. A kanonikus JSON sort_keys=True-pal készül, így a Python-szótárak kulcssorrendje nem változtatja meg a kulcsot.

Ha a feladat kódja úgy változik, hogy más eredményt ad ugyanarra a bemenetre, a verziószámot is emelni kell. A task_version="v1" szándékosan explicit: a kódból automatikusan kinyert hash is használható, de a kézi verzió átláthatóbb és a laborban könnyen ellenőrizhető.

30.3.3 Retry, scheduling és backfill

A tranziens hibákat — rövid fájlzár, hálózati időtúllépés — érdemes újrapróbálni. Az újrapróbálkozás (retry) csak determinisztikus, idempotens feladatra biztonságos: ha ugyanaz a hívás kétszer fut le, ugyanazt az outputot kell adnia. Példánkban a CSV-olvasás ilyen.

Az ütemezés (scheduling) azt jelenti, hogy egy külső folyamat időzítve indítja el a teljes DAG-futtatást, például naponta új adatpartíció beérkezése után. A cache és az ütemezés független: az ütemező minden nap futtathatja a DAG-ot, a cache pedig megakadályozza, hogy változatlan bemenetre újra számoljunk.

A visszamenőleges feldolgozás (backfill) olyan korábbi logikai dátumokra vagy partíciókra futtatja a DAG-ot, amelyek kimaradtak, vagy amelyekhez javított bemenet érkezett. A Northstar időhelyes jelentése itt is érvényes: ha a label_available_at értékek utólag teljesebbé válnak, akkor a backfill más címkéket láthat. A backfill tehát nemcsak a hiányzó futások újraindítása, hanem a predikció pillanatának (lásd Adatéletciklus és a predikció pillanata és Tanítás, validáció és érintetlen teszt) explicit újrafelvétele is. Azt is meg kell őrizni, hogy a régi cache-kulcsok ne tűnjenek el: egy korábbi javítás előtti állapot visszaállításához a régi output még kellhet.

30.3.4 Teljes minimális helyi DAG

A következő szkript standard könyvtárat használ, ezért külön csomagtelepítés nélkül futtatható orders_mini.csv mellett. A fájlt a kód első futáskor létrehozza, így a példa önálló. Linux/macOS terminálban a parancs python cache_pipeline.py; Windows PowerShellben ugyanez. A projektben uv run python cache_pipeline.py is megfelel.

import csv
import hashlib
import json
import time
from pathlib import Path

CACHE_DIR = Path("cache")
CACHE_DIR.mkdir(exist_ok=True)
MAX_ATTEMPTS = 3
RETRY_WAIT_SECONDS = 0.5
TASK_VERSION = "v1"

mini_csv = "\n".join(
    [
        "order_id,customer_id,order_ts,quantity_sum,priority_flag",
        "o1,c1,2025-01-01T08:00:00+00:00,3,1",
        "o2,c1,2025-01-02T08:00:00+00:00,2,0",
        "o3,c2,2025-01-03T08:00:00+00:00,5,1",
        "o4,c2,2025-01-04T08:00:00+00:00,1,1",
        "o5,c1,2025-01-05T08:00:00+00:00,4,0",
        "o6,c3,2025-01-06T08:00:00+00:00,2,0",
    ]
)
Path("orders_mini.csv").write_text(mini_csv, encoding="utf-8")

def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()

def task_cache_key(task_name, config, input_hashes):
    payload = {
        "task_name": task_name,
        "task_version": TASK_VERSION,
        "config": config,
        "input_hashes": input_hashes,
    }
    canonical = json.dumps(payload, sort_keys=True).encode("utf-8")
    return sha256_bytes(canonical)

def read_rows(path):
    with open(path, newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))

def extract(input_files, config):
    return read_rows(input_files[0])

def filter_priority(input_files, config):
    rows = json.loads(Path(input_files[0]).read_text(encoding="utf-8"))
    want = config["only_priority"]
    return [r for r in rows if int(r["priority_flag"]) == want]

def aggregate(input_files, config):
    rows = json.loads(Path(input_files[0]).read_text(encoding="utf-8"))
    totals = {}
    for r in rows:
        totals[r["customer_id"]] = totals.get(r["customer_id"], 0) + int(r["quantity_sum"])
    return totals

def run_task(task_name, config, input_files, run_fn):
    input_hashes = {}
    for p in input_files:
        input_hashes[p.name] = sha256_bytes(Path(p).read_bytes())

    key = task_cache_key(task_name, config, input_hashes)
    out = CACHE_DIR / f"{task_name}_{key}.json"

    for attempt in range(1, MAX_ATTEMPTS + 1):
        if out.exists():
            print(f"CACHE HIT  {task_name} {key[:12]}")
            return out, key, True

        print(f"CACHE MISS {task_name} {key[:12]} (attempt {attempt})")
        try:
            result = run_fn(input_files, config)
        except (OSError, RuntimeError) as exc:
            if attempt == MAX_ATTEMPTS:
                raise
            print(f"RETRY {attempt}/{MAX_ATTEMPTS}: {exc}")
            time.sleep(RETRY_WAIT_SECONDS)
            continue

        tmp = out.with_suffix(".tmp")
        tmp.write_text(json.dumps(result, indent=2), encoding="utf-8")
        tmp.replace(out)
        return out, key, False

def load_json(path):
    return json.loads(Path(path).read_text(encoding="utf-8"))

def run_pipeline(config):
    raw = Path("orders_mini.csv")
    extract_out, extract_key, hit_extract = run_task("extract", {}, [raw], extract)
    filter_out, filter_key, hit_filter = run_task(
        "filter", config, [extract_out], filter_priority
    )
    aggregate_out, aggregate_key, hit_aggregate = run_task(
        "aggregate", {}, [filter_out], aggregate
    )
    return (
        load_json(aggregate_out),
        (extract_key, filter_key, aggregate_key),
        (hit_extract, hit_filter, hit_aggregate),
    )

if __name__ == "__main__":
    result1 = run_pipeline({"only_priority": 1})
    print("priority=1:", result1[0])

    result2 = run_pipeline({"only_priority": 1})
    print("ismételt priority=1 hit-jei:", result2[2])
    assert result2[0] == result1[0]

    result3 = run_pipeline({"only_priority": 0})
    print("priority=0:", result3[0])
    assert result3[0] == {"c1": 6, "c3": 2}

A kódot nem futtattuk e generáláskor, ezért a következő napló a futtatás szemléltető szerkezeti mintáját mutatja. A tényleges hash-rövidítések a saját gépeden más hexa karakterek lesznek, mert az SHA-256 a teljes bájtfolyamot kódolja; a HIT/MISS mintázat viszont a fenti adatra így várható:

CACHE MISS extract <extract_hash_prefix> (attempt 1)
CACHE MISS filter <filter_hash_prefix> (attempt 1)
CACHE MISS aggregate <aggregate_hash_prefix> (attempt 1)
priority=1: {'c1': 3, 'c2': 6}

CACHE HIT  extract <ugyanaz_mint_fent>
CACHE HIT  filter <ugyanaz_mint_fent>
CACHE HIT  aggregate <ugyanaz_mint_fent>
ismételt priority=1 hit-jei: (True, True, True)

CACHE HIT  extract <ugyanaz_mint_fent>
CACHE MISS filter <más_filter_hash>
CACHE MISS aggregate <más_aggregate_hash>
priority=0: {'c1': 6, 'c3': 2}

30.3.5 Eredmény értelmezése és cache-invariáns

A kézi kontroll egyezik: priority=1 esetén az o1, o3, o4 sorok maradnak, az ügyfelenkénti összeg c1-nél 3, c2-nél 6. priority=0 esetén o2, o5, o6 marad, c1-nél 2 + 4 = 6, c3-nál 2. A teljes hat sor mennyisége 17, és a két szűrt aggregátum értékkészlete együtt is 17; ez a jó invariáns arra, hogy a szűrés nem veszített duplikátumot.

A második futás mindhárom feladatnál HIT: ugyanaz a feladatverzió, paraméter és bemenethash. A harmadik futásnál az extract inputja nem változott, ezért HIT; a filter paramétere megváltozott, ezért MISS; a downstream aggregate bemenete a szűrt fájl hashével változott, ezért szintén MISS. A pontosan megadott érvénytelenítési szabály tehát: ha a task verziója, paraméter-kanonikus alakja vagy bármely bemenet tartalomhashje eltér, a kulcs új, és a task újrafut. Ugyanakkor a cache-ből vett kimenetre is érvényesek az Adatteszt és numerikus teszt adattesztjei; a gyorsítás nem helyettesíti az output séma- vagy invariáns-ellenőrzését.

Backfill esetén rögzíteni kell a forrásváltozatot és a feldolgozás időszeletét; az újrafuttatott részablak csak akkor egyezik a teljes újraszámítással, ha az előzményfüggőségeket is biztosítjuk.

30.3.6 Orchestrator áttekintés: Prefect, Dagster és a dbt szerepe

A fenti run_task keret kézzel mutatja meg, mit ad egy munkafolyamat-vezénylő (orchestrator): függőségi sorrend, paraméterkanonizálás, retry, naplózás és cache. Kis helyi kutatásban ez a kézi DAG bőven elég. Ha már ütemezés, UI, riasztás, párhuzamos ágak vagy több futtatókörnyezet kell, akkor érdemes valódi orchestratort bevezetni.

Eszköz Fő fogalom Mit vesz át? Mikor indokolt?
Prefect flow és task retry, ütemezés, run-obszerabilitás, cache ha a Python-DAG és a szerver nélküli indítás is fontos
Dagster asset és materialization adatasset-függőség, partition/backfill, I/O menedzsment ha adatassetek és backfillek a központi kérdés
dbt ref() modellfa SQL-transzformációk függőségi gráfja ha a feldolgozás SQL-ben definiált, de a retry/backfill ettől még orchestrátorra marad

A fenti keretben a Prefect task dekorátora a run_task hívását csomagolja, a Dagster @asset pedig a feladatok kimenetét nevesített adatassetként kezeli. A konkrét dekorátorargumentumokat és verziókat a telepített változat dokumentációjából kell ellenőrizni; ez a könyv nem futtatta a Prefect/Dagster munkafolyamatokat. A dbt a SQL-transzformáció és dbt helye alfejezetben kerül részletesebben elő; most annyi a határa, hogy a dbt modell-definíció nem cache-mechanizmus, hanem függőségi leírás.

30.3.7 Feladatok

  1. Értelmezés. A hibás kulcsgenerátor key = task_name + output_fájl_neve alakban dolgozik. Milyen inputváltozást nem vesz észre a mini CSV-n? Melyik hiányzó kulcsfeltételre vezethető vissza a hiba? Adj konkrét példát egy új sor hozzáadásával.

  2. Megvalósítás. Általánosítsd a filter_priority feladatot filter_rows néven úgy, hogy a config only_priority mellett min_quantity egész értéket is kapjon, és csak azokat a sorokat tartsa meg, ahol quantity_sum >= min_quantity. Futtasd a {"only_priority": 1, "min_quantity": 4} paraméterrel a mini CSV-n. Mi a várt aggregált kimenet? Ellenőrizd, hogy a min_quantity=0 változat a korábbi only_priority=1 cache-től eltérő kulcsot kap.

  3. Hibadiagnózis. A csapat a filter_priority függvényt úgy módosította, hogy a priority_flag egyezést kisebb-egyenlő relációra cserélte. A napló mégis CACHE HIT-et írt, és a régi kimenet jött vissza. Mit hagyott ki a kulcs, és mi a minimális javítás? Milyen adattesztet javasolsz Adatteszt és numerikus teszt alapján a javítás után?

  4. Megváltozott feltétel és backfill. Egészítsd ki a mini CSV-t a sorral o7,c2,2025-01-07T08:00:00+00:00,6,1. Futtasd újra {"only_priority": 1}-gyel. Milyen HIT/MISS mintázatot vársz a három taskra, és mi lesz az aggregátum? Maradnak-e meg a régi cache-fájlok? Fogald meg, miért fontos ez a régi kulcsok megtartása egy backfill visszaállításakor.

30.3.8 Megoldások és indoklás

  1. A hibás kulcs nem tartalmazza a bemeneti fájl tartalomhashét. Ha a mini CSV-be beszúrjuk a o7,c2,...,6,1 sort, a fájlnév ugyanaz marad, tehát a filter a régi cache-ből HIT-et adna, és az új o7 sor nem jelenne meg az aggregátumban. A hiányzó feltétel az input hash. A javított kulcs minden bemeneti fájl bájtjának SHA-256-jét beépíti, ezért az új tartalom új kulcsot ad.

  2. A filter_rows megvalósítása:

def filter_rows(input_files, config):
    rows = json.loads(Path(input_files[0]).read_text(encoding="utf-8"))
    kept = []
    for r in rows:
        ok_priority = int(r["priority_flag"]) == config["only_priority"]
        ok_quantity = int(r["quantity_sum"]) >= config["min_quantity"]
        if ok_priority and ok_quantity:
            kept.append(r)
    return kept

A {"only_priority": 1, "min_quantity": 4} paraméterrel az o1, o3 és o4 közül csak o3 marad, mert o1 quantity=3, o4 quantity=1; az aggregátum {"c2": 5}. A min_quantity=0 és only_priority=1 config kanonikus kulcsa eltér a min_quantity=4 változatétól, mert a numerikus paraméter is a config része. A gyors ellenőrzés: a két filter_ cache-fájl hexa rövidítése különbözik.

  1. A hiba a task verziójának kihagyása. Ugyanaz a feladatnév, paraméter és input hash mellett a régi cache kimenete érvényesnek látszik, pedig a kód új logikája más eredményt adna. A minimális javítás: TASK_VERSION = "v2"-re emelés, mert a függvény viselkedése megváltozott. Ezután a kulcs megváltozik, és a filter újrafut. A javítás után egy fixture-alapú teszt is indokolt: állandó, kis bemenetre a szűrő elvárt sorszámát és az aggregátumot ellenőrizd. A zöld teszt a szoftver szerződését igazolja, a cache nem.

  2. Az o7 hozzáadása után az extract input CSV bájtjai változnak, ezért extract MISS. A filter bemenete az új extract kimenet, így filter MISS; az aggregate downstream is MISS. A priority=1 sorok: o1 (3), o3 (5), o4 (1), o7 (6) → c1=3, c2=12. A várt aggregátum tehát {"c1":3,"c2":12}. A régi cache-fájlok megmaradnak, mert kulcsukban a régi input hash szerepel. Ez teszi lehetővé, hogy a backfillnél egy korábbi adatpartícióra érvényes kimenet megmaradjon, és a javítást követően is összehasonlíthassuk a két állapotot. A régi kulcsok megtartása tehát nem hulladék, hanem a visszamenőleges követés része.

30.3.9 Gyakori tévedés

A leggyakoribb hiba: „a cache-ben lévő output bizonyítja, hogy az aktuális bemenethez készült.” Ellenpélda a fenti 1. feladat: a filter_default.json fájl létezik, ezért a hibás kulcs HIT-et ad, holott a CSV közben új rendelést kapott. A HIT igazságtartalma csak akkor áll, ha a kulcs tartalmazza a bemenet tényleges hashét és a feladat verzióját; a fájlnév alapú cache ezt a feltételt nem teljesíti.

Rövid önellenőrzés: meg tudnád-e mondani egy cache/filter_*.json fájl nevéből, hogy melyik paraméterrel és melyik inputtartalommal készült? A válasz csak akkor igen, ha a hexa kulcsban a megfelelő hash szerepel, és ismered a kanonizálási szabályt; önmagában a fájlnév nem bizonyíték.

30.3.10 Rövid összefoglaló

A cache kulcs a feladat nevét, verzióját, a paraméterek kanonikus JSON-ját és minden bemenet SHA-256 hashét tartalmazza. Ha bármelyik változik, új kulcs jön létre, és a task újrafut; ha mindegyik azonos, a mentett output felhasználható. A retry idempotens, tranziens hibát kezel. A backfill korábbi logikai dátumokra vagy partíciókra futtat vissza; a régi kulcsok megtartása a visszaállítást és az összehasonlítást támogatja. A kézi DAG kis projektre elég; Prefect vagy Dagster akkor indokolt, ha ütemezés, UI és asset-megfigyelhetőség is kell. A következő SQL-transzformáció és dbt helye a SQL-transzformációk és a dbt modellfa helyét vizsgálja ugyanebben a csővezetékben.