A Northstar pipeline a nyers rendeléseket naponta betölti; a Feldolgozási gráf és idempotencia alfejezetben
definiált DAG azt ígéri, hogy a raw/ rétegből a
validated/ rétegbe csak olyan adat kerül, amelyre a
downstream feature- és modellépítés támaszkodhat. A mai egységben az a
kérdés, hogyan állítsunk fel olyan kaput, amely három hibás bemenetre
megbízhatóan a megfelelő döntést adja. A megfigyelési egység a rendelés,
a kapu bemenete egy napi orders_YYYYMMDD.parquet fájl és a
customers.parquet törzs, a kimenet pedig a döntés
(STOP, QUARANTINE, WARN), a
validált tábla és a karanténnapló. Induló becslés: a sémahibás fájlra a
kapunak blokkolnia kell; a rossz ügyfélhivatkozást tartalmazó fájlból a
jó sorokat tovább kell engednie, a rossz sort le kell választania; a
frissességi problémát pedig figyelmeztetéssel kell jeleznie. A
következőkben ezeket a döntési szabályokat pontosítjuk.
A minőségi szabályhoz rendelt döntés nem az adathiba érzelmi súlyát,
hanem a downstream kockázatát írja le. A STOP azt jelenti,
hogy a lépés nem ír új kimenetet, és a hívó pipeline megszakad. A
QUARANTINE a hibás sorokat külön táblába teszi, a megfelelő
sorokat továbbengedi; a kiválasztás nem csendes, mert a karanténnapló
rögzíti a sorszámot, a szabályt és az időbélyeget. A WARN
nem blokkol, de a futásjegyzékben emlékeztetőt hagy. A három szint
használata akkor korrekt, ha a szabályhoz tartozó hibahatás ismert.
| Szabály | Tipikus döntés | Indoklás |
|---|---|---|
| Schema drift | STOP | A downstream kód típusokra épül; a továbbengedés rejtett hibákat hozhat. |
| Soronkénti FK-sértés | QUARANTINE | A hibás sor lokális, a többi sor megbízható. |
| Enyhe freshness-késés | WARN | Az adat még használható, de a folyamat késését jelezni kell. |
A döntés mátrixa nem rögzíthető örökre: ha a frissességet a
predikcióhoz azonnal kell használni, a késés STOP is lehet. A
konfigurációban ezért szerepeljen a severity mező, ahogyan
a Konfiguráció és futási napló alfejezetben a
konfigurációt és a futásnaplót bevezettük.
A séma a bemeneti oszlopok nevének, típusának és nullázhatóságának
szerződése. A napi rendelésfájl rögzített sémája a
Northstar-szerződésből származik: order_id szöveg,
order_ts időbélyeg, distance_km lebegőpontos,
priority_flag egész, a többi oszlop a rendelés pillanatában
ismert. A schema drift azt jelenti, hogy a tényleges fájl
ettől eltér. Ha például a distance_km oszlop
'12.5' karakterláncként érkezik, a későbbi numerikus
művelet vagy hibát dob, vagy csendesen stringösszehasonlítást végez. A
kapu a DuckDB DESCRIBE kimenetét hasonlítja a rögzített
sémához. A pontos típusnevek verziófüggők lehetnek, ezért éles használat
előtt egy helyi DESCRIBE futással kell a várt értékeket
rögzíteni.
A sorok száma a fájl teljességének durva, de olcsó ellenőrzése. Egy
üres fájl vagy a duplázott betöltés azonnal látszik. A napi forgalomra a
korábbi manifestekből származó alsó és felső küszöböt adunk meg: például
lo=1, hi=500. Ha a tényleges sorok száma ezen
kívül esik, a kapu STOP-pal áll meg, mert a fájl nem a várt napi adag. A
küszöböt nem szabad a mintából egyetlen nap alapján becsülni; promóció
vagy ünnep miatt a tartomány szűk lehet, és a kapu hibásan blokkol.
Ilyenkor a konfigurációban egy row_count_override mező és a
manifestbe írt indoklás a helyes megoldás.
A frissesség azt méri, mennyire aktuális a fájl maximális időbélyege
a kapu futásának idejéhez képest:
.
Ha a fájl tegnap érkezett, de a benne lévő rendelések 5 napja vannak, a
betöltési folyamat leállt vagy késik. A szabály
age > limit esetén WARN-et ad, kivéve, ha a modell az
aznapi döntéshez használja az adatot; akkor a szabály STOP is lehet. A
freshness nem az adattartalom hibája, hanem a folyamaté; a Konfiguráció és futási napló futásjegyzékben a
késést külön mezőben tároljuk.
A completeness oszloponként a nem hiányzó értékek aránya:
.
Ha 120 sorból 3 hiányzó distance_km van, az arány 0,975,
azaz 97,5%. A szabály nem csak az arányt, hanem az abszolút hibaszámot
is nézze: egy 10 000 soros fájlban a 0,5% hiány 50 sor, ami egy
modellben már eltérő döntést hozhat. A kapu a kötelező prediktor
oszlopokra missing_count = 0 szabályt ír elő, és a hibás
sorokat karanténba teszi. A hiányzási mechanizmus (MCAR, MAR, MNAR)
dönti el, hogy a karantén milyen torzítást visz be; a Randomizálás és kísérleti protokoll alfejezetben
látott mintaszelekció ugyanazt a kérdést veti fel: a kihagyott sorok nem
véletlenek, ezért a validált eloszlás nem feltétlenül a teljes
forgalomé. A kapu nem imputál, csak szeparál.
Az idegen kulcs ellenőrzése azt jelenti, hogy minden
customer_id értéknek léteznie kell a customers
törzsben. Az ellenőrzés anti-join: LEFT JOIN a törzshöz, és
ahol customers.customer_id IS NULL, ott orphan sor van. Ha
az ügyfél új, és a törzs még nem frissült, a helyes lépés a törzs
feltöltése, nem az FK megsemmisítése. A kapu a rossz hivatkozású sorokat
karanténba teszi.
A példa a projektgyökérben fut, a lab/ könyvtárban. A
customers.parquet a Northstar 300 ügyfelét tartalmazza. A
napi rendelésfájl sémája a fent rögzített. A kapu Python-szkriptje
DuckDB-t használ, mert a Parquet-olvasás és az anti-join egyaránt
SQL-ben kifejezhető.
# quality_gate_daily.py (szemléltető; futtatás előtt a DuckDB DESCRIBE
# kimenetét a helyi verzióhoz igazítsd)
import json
import sys
from datetime import timedelta
from pathlib import Path
import duckdb
EXPECTED_SCHEMA = {
"order_id": "VARCHAR",
"customer_id": "VARCHAR",
"order_ts": "TIMESTAMP WITH TIME ZONE",
"promised_days": "BIGINT",
"quantity_sum": "BIGINT",
"item_rows": "BIGINT",
"basket_minor": "BIGINT",
"shipping_fee_minor": "BIGINT",
"currency": "VARCHAR",
"distance_km": "DOUBLE",
"priority_flag": "BIGINT",
}
def check_schema(con, table):
actual = {r[0]: r[1].upper() for r in con.execute(f"DESCRIBE {table}").fetchall()}
problems = []
for col, dtype in EXPECTED_SCHEMA.items():
if col not in actual:
problems.append(f"missing column {col}")
elif actual[col] != dtype:
problems.append(f"{col}: expected {dtype}, got {actual[col]}")
return problems
def check_row_count(con, table, lo, hi):
n = con.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
if n < lo or n > hi:
return [f"row count {n} outside [{lo}, {hi}]"]
return []
def check_completeness(con, table, required, max_missing=0):
out = []
for col in required:
m = con.execute(f"SELECT COUNT(*) FROM {table} WHERE {col} IS NULL").fetchone()[0]
if m > max_missing:
out.append(f"{col}: {m} missing")
return out
def check_freshness(con, table, now_ts, limit):
max_ts = con.execute(f"SELECT MAX(order_ts) FROM {table}").fetchone()[0]
age = now_ts - max_ts
if age > limit:
return [f"freshness age {age} > {limit}"]
return []
def check_referential_integrity(con, orders_table, customers_table, fk_col):
orphans = con.execute(f"""
SELECT COUNT(*) FROM {orders_table} o
LEFT JOIN {customers_table} c USING ({fk_col})
WHERE c.{fk_col} IS NULL
""").fetchone()[0]
if orphans:
return [f"{orphans} orphan {fk_col} rows"]
return []
def decide(schema, rows, completeness, freshness, referential):
decision = "WARN"
blockers = []
if schema:
decision = "STOP"; blockers += schema
if rows:
decision = "STOP"; blockers += rows
if completeness:
decision = "QUARANTINE"; blockers += completeness
if referential:
decision = "QUARANTINE"; blockers += referential
if freshness:
blockers.append("WARN")
return decision, blockersA három hibás bemenet előállítása külön szkriptben:
# make_three_bad_inputs.py (szemléltető)
import duckdb
con = duckdb.connect()
# 1. Sémahiba: distance_km VARCHAR
con.execute("""
CREATE TABLE good AS
SELECT 'o1' AS order_id, 'c1' AS customer_id,
TIMESTAMPTZ '2025-04-20 10:00+00' AS order_ts,
7 AS promised_days, 2 AS quantity_sum, 1 AS item_rows,
1200 AS basket_minor, 300 AS shipping_fee_minor,
'EUR' AS currency, 8.5 AS distance_km, 0 AS priority_flag
""")
con.execute("ALTER TABLE good ALTER distance_km TYPE VARCHAR")
con.execute("COPY good TO 'orders_bad_schema.parquet' (FORMAT PARQUET)")
# 2. Idegen kulcs hiba: 11 jó sor + 1 ismeretlen ügyfél
con.execute("DROP TABLE good")
con.execute("""
CREATE TABLE good AS
SELECT 'o1' AS order_id, 'c1' AS customer_id,
TIMESTAMPTZ '2025-04-20 10:00+00' AS order_ts,
7 AS promised_days, 2 AS quantity_sum, 1 AS item_rows,
1200 AS basket_minor, 300 AS shipping_fee_minor,
'EUR' AS currency, 8.5 AS distance_km, 0 AS priority_flag
""")
con.execute("""
INSERT INTO good VALUES
('o2', 'c1', TIMESTAMPTZ '2025-04-20 11:00+00', 7, 3, 2, 1500, 300, 'EUR', 5.1, 0),
('o3', 'c1', TIMESTAMPTZ '2025-04-20 12:00+00', 7, 1, 1, 800, 300, 'EUR', 2.0, 0),
('o4', 'c1', TIMESTAMPTZ '2025-04-20 13:00+00', 7, 1, 1, 900, 300, 'EUR', 3.2, 0),
('o5', 'c1', TIMESTAMPTZ '2025-04-20 14:00+00', 7, 4, 3, 2200, 300, 'EUR', 6.0, 0),
('o6', 'c1', TIMESTAMPTZ '2025-04-20 15:00+00', 7, 2, 1, 1100, 300, 'EUR', 7.0, 0),
('o7', 'c1', TIMESTAMPTZ '2025-04-20 16:00+00', 7, 1, 1, 500, 300, 'EUR', 1.5, 0),
('o8', 'c1', TIMESTAMPTZ '2025-04-20 17:00+00', 7, 1, 1, 700, 300, 'EUR', 2.2, 0),
('o9', 'c1', TIMESTAMPTZ '2025-04-20 18:00+00', 7, 2, 2, 1300, 300, 'EUR', 4.0, 0),
('o10', 'c1', TIMESTAMPTZ '2025-04-20 19:00+00', 7, 3, 2, 1800, 300, 'EUR', 3.8, 0),
('o11', 'c1', TIMESTAMPTZ '2025-04-20 20:00+00', 7, 2, 1, 950, 300, 'EUR', 5.5, 0),
('o12', 'C-999-UNKNOWN', TIMESTAMPTZ '2025-04-20 21:00+00', 7, 1, 1, 600, 300, 'EUR', 4.4, 0)
""")
con.execute("COPY (SELECT * FROM good) TO 'orders_bad_fk.parquet' (FORMAT PARQUET)")
# 3. Frissesség hiba: rendelések 13 napja
con.execute("""
CREATE TABLE stale AS
SELECT * FROM good
WHERE order_ts < TIMESTAMPTZ '2025-04-21'
""")
con.execute("COPY stale TO 'orders_bad_freshness.parquet' (FORMAT PARQUET)")A kapu futtatása (Linux/macOS terminál, projektgyökér munkakönyvtár):
python quality_gate_daily.py orders_bad_schema.parquet customers.parquet --now "2025-05-03T00:00:00+00:00"
python quality_gate_daily.py orders_bad_fk.parquet customers.parquet --now "2025-05-03T00:00:00+00:00"
python quality_gate_daily.py orders_bad_freshness.parquet customers.parquet --now "2025-05-03T00:00:00+00:00"Az elvárt kimenet kézi ellenőrzéssel, nem futtatott kód alapján:
orders_bad_schema.parquet: a DESCRIBE a
distance_km-re VARCHAR-t ad, ezért
STOP, exit code 2, nincs validált kimenet.orders_bad_fk.parquet: az anti-join 1 orphansort talál,
ezért QUARANTINE; a 11 jó sor a
validated_orders_daily táblába, az o12 sor a
quarantine_orders táblába kerül. A manifestben
quarantined_rows=1.orders_bad_freshness.parquet: a maximális
order_ts 2025-04-20, a now_ts 2025-05-03, így
age=13 nap, ami nagyobb, mint a limit=2 nap; a
döntés WARN, a fájl továbbmegy, de a manifest
warnings mezője tartalmazza a frissesség eltérést.A dbt-ben ugyanezek a szabályok adatmodell-yml-ben élnek, és a
dbt build futtatásakor automatikusan ellenőrződnek. A
relationships teszt a referential integrity-t, a
not_null a completeness-t, a accepted_values a
megengedett kategóriákat ellenőrzi. A sorok számát és a frissességet a
dbt test custom tesztek vagy a
source freshness fogalmával lehet mérni. A dbt szerepe
ebben a fejezetben áttekintés: a Python-kapu és a dbt-teszt nem zárja ki
egymást, a dbt a tárolt adatmodell szerződését, a Python-kapu pedig a
fájlbetöltés előtti ellenőrzést végzi. Prefect vagy Dagster akkor
indokolt, ha a kapu után retry, ütemezés vagy több lépésből álló
megfigyelhetőség is kell.
A napi fájlban a distance_km 3 sora NULL. Miért nem
fogadható el az, hogy a kapu egyszerűen 0-ra cseréli ezeket, és
WARN döntést ad? Fogalmazza meg a hiányzási mechanizmus és
a modellkalibráció szempontjából.
Írja át a check_referential_integrity függvényt úgy,
hogy a hibás customer_id értékek listáját is visszaadja, ne
csak a darabszámot. A mini bemenet: orders sorai
('o1','c1'), ('o2','c2'), ('o3','cX');
customers sorai 'c1','c2'. Adja meg a várt
kimenetet.
Egy napi kapu lo=1, hi=500 értékkel fut, és a fájl 6000
sort tartalmaz. A downstream csapat szerint a fájl helyes, egy promóció
miatt nőtt meg a forgalom, nincs duplikáció. A kapu STOP-ot ad. Mi a
tanulság, és hogyan javítható a konfiguráció anélkül, hogy a kaput
kikapcsolnánk?
Írjon kaput az assignments.parquet és
outcomes.parquet kísérleti pároshoz. A megfigyelési egység
a randomizált ügyfél. Adjon legalább öt szabályt, mindegyikhez
súlyosságot, és egy érvényességi korlátot. A szabályok között
szerepeljen: customer_id egyedi az assignmentsben;
variant csak control vagy
treatment; outcomes customer_id halmaza
pontosan az assignments; conversion_7d értéke {0,1};
observed_until = 2025-07-08.
A nulla imputálás azt állítja, hogy a rendelés nulla kilométerről érkezett, ami a szállítási idő modelljében a lehető legrövidebb távolságnak felel meg. Ha a hiány nem teljesen véletlen (MCAR), hanem például a távolságot nem rögzítő külső partner rendszeréből származik (MAR/MNAR), akkor a hiányzó csoport nem reprezentálja a teljes eloszlást. A Kalibrált valószínűség és predikciós bizonytalanság alfejezetben látott kalibrációhoz a prediktor eloszlásának torzulása rossz bizonytalansági becslést ad. A helyes döntés a sorok karanténba tétele, és a hiány okának vizsgálata; ha imputálás szükséges, az külön elemzési lépés, a Konfiguráció és futási napló manifestben rögzített módon.
def check_referential_integrity(con, orders_table, customers_table, fk_col):
orphans = con.execute(f"""
SELECT DISTINCT o.{fk_col}
FROM {orders_table} o
LEFT JOIN {customers_table} c USING ({fk_col})
WHERE c.{fk_col} IS NULL
ORDER BY 1
""").fetchall()
if orphans:
ids = [r[0] for r in orphans]
return [f"{len(ids)} orphan rows: {ids}"]
return []A mini bemenetre az anti-join egy sort ad: cX. Az
ellenőrző invariant: a visszaadott lista hossza egyenlő a
COUNT(*)-tal, és a lista minden eleme valóban hiányzik a
customers-ből. A tolerancia itt 0: a feladat egzakt
string-összehasonlítás.
A hiba nem a fájlban, hanem a kapu konfigurációjában van. A
hi=500 küszöb egy régi átlagos napra volt kalibrálva; a
promóció ismert esemény, ezért a konfigurációban a
row_count_upper_override=7000 és az indoklás kerüljön a
manifestbe. A kaput nem szabad kikapcsolni, mert a duplikáció elleni
védelem továbbra is kell; a küszöböt a promóció várható hatásával kell
frissíteni, és a folyamatot utólag ellenőrizni, hogy a 6000 sor tényleg
a várt növekedés-e.
A kísérleti kapu szabályai:
assignments.customer_id egyedi: ha nem, STOP, mert a
randomizálási egység sérült.assignments.variant értékkészlete {control, treatment}:
ha más, QUARANTINE a hibás sorokra.outcomes.customer_id halmaza pontosan az assignments:
ha extra ügyfél van, QUARANTINE; ha hiányzik, STOP, mert az ITT nevező
nem teljes (visszautalás Randomizálás és kísérleti
protokoll).outcomes.conversion_7d értéke {0,1}: ha más,
QUARANTINE.outcomes.observed_until = 2025-07-08: ha eltér, STOP,
mert a megfigyelési ablak nem azonos.outcomes.currency = EUR: ha eltér, WARN, mert a
pénzösszeg értelmezése változhat.Az érvényességi korlát: a kapu az adat formális szerződését ellenőrzi, nem a kísérleti randomizálás tényleges végrehajtását; a randomizálás minőségét külön egyensúlyteszttel kell vizsgálni.
A leggyakoribb tévhit az, hogy a kapu célja a nullák eltüntetése. Ha
a distance_km hiányzó értékét nulla kilométerre írjuk, a
modell azt tanulja, hogy a hiányos rekordok gyors kézbesítésűek; egy
logisztikai döntésnél ez szisztematikusan alulbecsli a késést. A Kalibrált valószínűség és predikciós bizonytalanság
kalibrációs görbéje ilyenkor nem a valós gyakoriságot mutatja, hanem a
mesterséges nulla távolság hatását. A helyes stratégia először a sorok
karanténba tétele, majd a hiány okának vizsgálata; imputálás csak akkor,
ha a mechanizmust és a torzítást explicit módon leírtuk. A Konfiguráció és futási napló futásnaplóban ennek a
döntésnek nyoma kell legyen.
A hibás rekordok kezelését hibakategóriákkal és karanténkimenettel dokumentáljuk; a csöndes eldobás a célpopulációt és a becslést is megváltoztathatja.
A minőségi kapu minden szabályhoz STOP, QUARANTINE vagy WARN döntést rendel, és a döntést a manifestben rögzíti. A schema drift és a durva row-count eltérés blokkol, a soronkénti hiány és idegenkulcs-sértés karanténba kerül, a frissesség késése figyelmeztetést ad. A karantén nem imputálás; a hiányzási mechanizmus torzítást vihet a későbbi modellbe, ezért a döntést indokolni és naplózni kell. A Python-kapu a fájlbetöltés előtti ellenőrzést végzi, a dbt a tárolt modell szerződését teszteli, a Prefect/Dagster pedig akkor indokolt, ha retry és ütemezés is kell. A következő alfejezetben a cache, backfill és orchestrator kérdését vesszük elő.