6.3 Duplikáció, kulcs és adatminőségi szerződés

6.3.1 Cél és konkrét helyzet

A Northstar kezdőlabor orders_raw.csv fájlja szándékosan hibás másolata a tiszta rendelési adatnak. A megfigyelési egység itt egy rendelés, a sorazonosító pedig az order_id. A tisztaz orders.csv szerződése szerint minden order_id pontosan egyszer fordul elő, és a fájl 2400 sort tartalmaz. A nyers fájl ezzel szemben 2405 soros: öt extra sor pontos, teljes sorra kiterjedő másolat, ezen felül más hibák is vannak benne, amelyeket későbbi alfejezetek kezelnek.

Ebben a laborban az a feladat, hogy a duplikált sorokat dokumentált szabállyal eltávolítsuk, a kivett sorokat karanténba helyezzük, és a tisztított darabszámot a kontrollhoz hasonlítsuk. A kód megírása előtt adjunk előzetes becslést: öt sort kell karanténba tenni, és 2400 sornak kell maradnia. Ha a futás ettől eltér, akkor vagy a fájl nem a szerződés szerinti verzió, vagy a szabály hibás.

6.3.2 Fogalmi magyarázat

6.3.2.1 Megfigyelési egység és sorazonosító

Egy tábla akkor „tiszta” ebből a szempontból, ha a megfigyelési egység és a kulcs között kölcsönösen egyértelmű kapcsolat van:

$$N_{\text{sor}} = N_{\text{egyedi rendelés}} + N_{\text{extra másolat}}$$

ahol $N_{\text{sor}}$ a fájl sorainak száma, $N_{\text{egyedi rendelés}}$ a különböző order_id értékek száma, $N_{\text{extra másolat}}$ pedig a duplikátumok miatt felesleges sorok száma. Ha nincs kulcsütközés és hiányzó kulcs, a laborban ez $2405 = 2400 + 5$.

6.3.2.2 duplicated és drop_duplicates

A DataFrame.duplicated() (magyarul: ismétlődő) logikai maszkot ad: True, ha az adott sor a korábbi sorok pontos másolata. Az alapértelmezés minden oszlopot vizsgál, és a keep="first" az első előfordulást nem jelöli duplikátumnak. A drop_duplicates() ezt a maszkot használja a sorok eldobására.

import pandas as pd

mini = pd.DataFrame({
    "order_id": ["o1", "o2", "o1"],
    "amount": [10, 20, 10],
})

mini.duplicated(keep=False).tolist()
# [True, False, True]

A subset=["order_id"] paraméterrel csak a kulcsra szűrünk. Ez veszélyesen hallgat el egy fontos esetet: ha ugyanaz a kulcs kétszer szerepel, de a többi oszlop eltér, az kulcsütközés, nem pedig egyszerű másolat. Ilyenkor az első vagy utolsó sor automatikus megtartása adatvesztést okozhat, ezért a pontos másolat tényét külön ellenőrizni kell.

6.3.2.3 Egyediség és nullarány

Az egyediség (uniqueness) ellenőrzése a tisztítás után:

clean["order_id"].is_unique

Ennek megbízható használatához a kulcsban nem lehet hiányzó érték. A nullarány (missing rate) képlete egy oszlopra:

$$\text{nullarány}(x) = \frac{1}{n}\sum_{i=1}^{n} \mathbb{1}[\text{hiányzó}(x_i)]$$

ahol $n$ a sorok száma, $x_i$ az $i$-edik sor értéke, és $\mathbb{1}[\cdot]$ 1, ha az érték hiányzik, 0 különben. A mutató értéke 0 és 1 közé esik. A pandas a NaN értékeket a duplikációvizsgálatban egyenlőnek tekinti, ezért két különböző, hiányzó order_id-jű sor hamisan duplikátumnak látszhat. A kulcs nullarányát tehát a deduplikálás előtt ellenőrizni kell.

6.3.2.4 Karantén és audit trail

A karantén (quarantine) olyan külön táblába mentett gyűjtemény, amely az eldobott sorokat és az eldobás okát tartalmazza. Az audit trail (auditnapló, ellenőrzési nyomvonal) a tisztítási lépés bemenetét, szabályát és kimeneti darabszámait rögzíti. Így a döntés később visszakereshető, és a folyamat újra lejátszható.

6.3.3 Teljes minimális munkapélda

A következő szkriptet a projektgyökérből futtatjuk. Platformfüggetlen pathlib útvonalat használunk. Az időbélyeget most szándékosan szövegként olvassuk be: a duplikációvizsgálat a fájl tartalmát hasonlítja, a dátumkonverziót a Dátumok, időzónák és kategóriák alfejezetre hagyjuk.

from pathlib import Path
import json

import pandas as pd

DATA_DIR = Path("lab")
OUT_DIR = Path("work/s023")
OUT_DIR.mkdir(parents=True, exist_ok=True)

raw = pd.read_csv(
    DATA_DIR / "orders_raw.csv",
    dtype={
        "order_id": "string",
        "customer_id": "string",
        "currency": "string",
    },
)

# 1. A kulcs legyen teljesen kitöltött; NaN miatt hamis duplikátum lehet.
key_null_rate = float(raw["order_id"].isna().mean())
assert key_null_rate == 0.0, f"order_id hiányzik: {key_null_rate:.4f}"

# 2. Teljes sorra kiterjedő duplikátumok.
full_dup_mask = raw.duplicated(keep=False)
full_dup_df = raw.loc[full_dup_mask].copy()

# 3. Ellenőrzés: a duplikált sorok valóban pontos másolatok-e kulcsonként.
payload_wide = full_dup_df.groupby("order_id", dropna=False).nunique(dropna=False)
payload_identical = (payload_wide == 1).all(axis=1)
assert payload_identical.all(), "Kulcsütközés: eltérő tartalmú sorok azonos order_id-vel."

dup_counts = full_dup_df.groupby("order_id", dropna=False).size()
assert dup_counts.max() == 2, "Ebben a laborban csak páros duplikátumokat várunk."

# 4. Egy másolat megtartása, a többi karanténba kerül.
to_remove_mask = raw.duplicated(keep="first")

quarantine = raw.loc[to_remove_mask].copy()
quarantine.insert(0, "audit_reason", "exact_full_row_duplicate")
quarantine.insert(1, "audit_rule", "keep_one_of_verified_exact_copies")

clean = raw.loc[~to_remove_mask].copy()

# 5. A sorazonosító egyedisége a kimenetben.
assert clean["order_id"].is_unique

# 6. Darabszám-összevetés a tiszta kontrollal.
control = pd.read_csv(
    DATA_DIR / "orders.csv",
    usecols=["order_id"],
    dtype={"order_id": "string"},
)
assert len(clean) == len(control)

# 7. Karantén és audit trail mentése.
quarantine.to_csv(OUT_DIR / "duplicate_quarantine.csv", index=False)
clean.to_csv(OUT_DIR / "orders_deduped.csv", index=False)

audit = {
    "input_file": "lab/orders_raw.csv",
    "input_rows": int(len(raw)),
    "duplicate_group_rows": int(full_dup_mask.sum()),
    "duplicate_groups": int(dup_counts.shape[0]),
    "rows_removed": int(to_remove_mask.sum()),
    "rows_kept": int(len(clean)),
    "control_rows": int(len(control)),
    "key_null_rate": key_null_rate,
    "rule": "keep_one_of_verified_exact_copies",
    "note": "A többi raw hibát a későbbi alfejezetek kezelik; ez a lépés csak a duplikációt javítja.",
}

(OUT_DIR / "audit.json").write_text(
    json.dumps(audit, indent=2, ensure_ascii=False),
    encoding="utf-8",
)

print(audit)

Futtatás uv-projektben:

# Windows, macOS és Linux shellben egyaránt
uv run python scripts/s023_dedup.py

A seed=20261020 labor szerződéséből levezethető, szemléltető értékek: input_rows=2405, duplicate_group_rows=10, duplicate_groups=5, rows_removed=5, rows_kept=2400, control_rows=2400. Ez nem helyi mérési eredmény; akkor válik mérté, ha a te gépeden a lab fájljaival lefut, és a manifest.json szerint azonosak a generálási paraméterek.

6.3.4 Eredmény értelmezése

A kód három invariant ellenőriz, mielőtt eldobna bármit: a kulcs nullaránya 0, minden duplikált csoport mérete 2, és a csoporton belüli tartalom minden oszlopban azonos. Ezért az öt eltávolított sor valóban felesleges másolat, nem pedig eltérő állapotú rendelés. A keep="first" itt csak azért biztonságos, mert a másolatok tartalma azonos; a fájlbeli sorrend nem hordoz üzleti jelentést.

A kimenet 2400 sora megegyezik a kontroll orders.csv darabszámával. Ez azonban csak darabszám-egyezés: a nyers fájlban még megvan a két hiányzó distance_km, a negatív basket_minor és a nem létező customer_id. A teljes tartalmi egyezéshez ezeket külön szabályokkal kell majd kezelni, és csak utána érdemes a kontrollhoz hasonlítani. A karanténfájl a kivett sorokat pontosan megőrzi, az audit.json pedig a lépés visszakereshetőségét biztosítja.

6.3.5 Feladatok

1. Értelmezés. Adott a következő kis tábla:

df = pd.DataFrame({
    "order_id": ["o1", "o2", "o1", "o3", "o3"],
    "amount":   [10,   20,   10,   30,   31],
})

Add meg a df.duplicated() és a df.duplicated(subset=["order_id"], keep=False) kimenetét. Hány sort tartana meg a df.drop_duplicates(subset=["order_id"], keep="first"), és milyen problémát rejt el?

2. Implementálás. Írj függvényt exact_duplicate_groups(df, key_column) néven, amely visszaad egy Series-t: azokra a kulcsokra True, ahol legalább két teljes sor duplikátum van, és a csoporton belül minden oszlop értéke azonos; különben False. A hiányzó kulcsokat ne kezeld automatikusan egy csoportként; a függvény dobjon ValueError-t, ha a kulcsban hiányzó érték van. Használd a groupby(...).nunique(dropna=False) technikát, és ellenőrizd az 1. feladat df-jén, ahol a várt eredmény az o1 kulcsra True, az o3 kulcsra False.

3. Hibadiagnózis. Egy kolléga a következőt futtatta:

deduped = updates.drop_duplicates(subset=["order_id"], keep="first")

A bemenet frissítési napló:

updates = pd.DataFrame({
    "order_id":    ["R1", "R2", "R1", "R2"],
    "captured_at": ["2025-01-01 08:00", "2025-01-01 08:00",
                    "2025-01-02 09:00", "2025-01-02 09:00"],
    "status":      ["new", "new", "shipped", "cancelled"],
})

Mi a szemantikai hiba a keep="first" használatában? Adj olyan dokumentált szabályt, amely megőrzi a legfrissebb állapotot, és mondd meg azt is, mikor kell karanténba tenni a csoportot dobás helyett.

4. Megváltozott feltétel. Egy új betöltés az eredeti 2405 soros orders_raw.csv mellett két újabb sort is tartalmaz ugyanazzal az order_id="X" kulccsal, de basket_minor értékük 100, illetve 200. Minden más megegyezik. Hány sort távolít el a fenti minimális példa szabálya, és hány sor marad? Miért bukik el az assert clean["order_id"].is_unique? Írd le a módosított szabályt, és add meg a várható kimeneti sorok számát is.

6.3.6 Megoldások és indoklás

1. A teljes sorra vonatkozó duplikáció maszkja [False, False, True, False, False], mert csak a harmadik sor pontos másolata az elsőnek. A kulcsra vonatkozó keep=False maszkja [True, False, True, True, True], mert az o1 és az o3 kulcs is többször szerepel. A drop_duplicates(subset=["order_id"], keep="first") három sort tart meg: o1 az amount=10, o2 az amount=20, o3 az amount=30 értékkel. Ez elrejti, hogy az o3 kulcshoz két különböző összeg tartozik; a 31 értékű sor csendesen eltűnik. Ha a szerződés szerint az o3 két állapota fontos, a csendes eldobás hibás adathoz vezet.

2. Egy megoldás:

def exact_duplicate_groups(df, key_column):
    if df[key_column].isna().any():
        raise ValueError("A kulcs nem tartalmazhat hiányzó értéket.")

    dup_mask = df.duplicated(keep=False)
    dup_df = df.loc[dup_mask]

    if dup_df.empty:
        return pd.Series(dtype=bool)

    payload_counts = dup_df.groupby(key_column).nunique(dropna=False)
    return (payload_counts == 1).all(axis=1)

A 3. feladat updates tábláján a függvény az R1 kulcsra False-t adna, mert a status és captured_at oszlopban eltérő értékek vannak a két sorban. A függvény nem dob el semmit, csak osztályoz: a pontos másolatokat elkülöníti a kulcsütközéstől.

3. A keep="first" azt feltételezi, hogy a fájlbeli sorrend üzletileg is az elsődleges sor. Itt viszont a captured_at az irányadó: a későbbi sor a frissebb állapot. Helyes szabály lehet: rendezés captured_at szerint, majd keep="last", de csak azonos order_id mellett, és miután ellenőriztük, hogy nincs két különböző állapot azonos időbélyeggel. Ha ugyanazon rendeléshez ugyanazon időpontban ellentmondó állapotok tartoznak, a csoportot nem szabad csendesen eldobni; karanténba kell tenni a teljes csoportot, és a forrásrendszer tulajdonosával egyeztetni.

4. Az eredeti példa szabálya a teljes sorra vonatkozó duplicated(keep=False) maszkot használja. Az X két sora nem teljes sorra azonos, mert a basket_minor különbözik, ezért a szabály csak az eredeti öt másolatot dobja el. A kimenet így $2400 + 2 = 2402$ sor lenne, és az assert clean["order_id"].is_unique megbukik, mert az X kétszer szerepel. Módosított szabály: először az exakt másolatokat távolítsuk el, majd a megmaradt kulcsütközéseket teljes csoportként karanténba tesszük. Az új, 2407 soros bemenetből az öt egzakt duplikátum eltávolítása után 2402 sor marad; ebből az X kulcsú két ütköző sor karanténba kerül, így a végeredmény 2400 soros. A kontrollal egyező sorSZÁM önmagában nem elegendő: az order_id-halmazt, a hibajegyzéket és a tartalmat is egyeztetni kell.

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

A leggyakoribb hiba a drop_duplicates(subset=["order_id"]) automatikus használata az üzleti szabály kimondása nélkül. Ha a duplikált kulcshoz eltérő tartalom tartozik, az első sor megtartása tetszőleges, és elrejti az adatminőségi problémát. Ezért a dobás előtt mindig kérdezzük meg: a megfigyelési egység és a kulcs tényleg ugyanaz? A kulcs hiánytalan-e? A duplikátum pontos másolat vagy kulcsütközés? A megtartott sor kiválasztása determinisztikus és dokumentált? A karanténba tett sorokat tudjuk-e később visszaállítani? A Lebegőpont, hiányzó érték és stabil számítás alfejezetből idézett numerikus óvatosság ide is érvényes: számított lebegőpontos oszlopoknál a ==-egyenlőség törékeny, ezért duplikációvizsgálat előtt a nyers fájl értékeit hasonlítsuk, vagy toleranciával ellenőrizzünk. A Git adat- és modellprojektekben Git-szabályai szerint a tisztító szkriptet és a szabályt verziózzuk; a generálható nagy nyers fájlok helyett a reprodukálhatósághoz elegendő a seed és a manifest hivatkozása.

Az egzakt duplikátumok eltávolítása elkülönítendő az azonos kulcsú, de eltérő adatokat tartalmazó rekordok feloldásától. Utóbbiakra a keep="first" nem általános javítási szabály; szükség van dokumentált üzleti döntésre vagy karanténra.

6.3.8 Rövid összefoglaló

A duplikációvizsgálat előtt tisztázni kell, hogy egy sor milyen megfigyelési egységet jelent. A duplicated teljes sorra, illetve subset paraméterrel kulcsra is vizsgálható; a kulcs szerinti automatikus eldobás azonban elrejtheti a kulcsütközést. A Northstar laborban az öt extra sor pontos másolat, ezért ellenőrzött csoportazonosság után egy példány megtartható, a többi karanténba kerül. A kimenet 2400 sora megegyezik a kontroll darabszámával, de a többi nyers hiba miatt ez még nem teljes tartalmi egyezés. Az audit trail rögzíti a szabályt, a bemeneti és kimeneti darabszámokat. A következő lépésben a dátumokat és időzónákat kezeljük ugyanezen a rendelési táblán.