8.1 CSV-től Parquet-ig: séma és partíció

8.1.1 Cél és konkrét helyzet

A Paraméterezett lekérdezés és lekérdezési terv alfejezetben SQL-lel kinyertük a Northstar-rendelések használható oszlopait. Most ugyanezt az adatterméket kell megosztani a kutatási notebookok és a későbbi batch-folyamatok között. A nyers orders.csv szöveges: nincs benne séma, ezért minden betöltő újra kitalálhatja a típusokat, és egy rosszul idézett mező csendben eltolhatja az oszlopokat.

A feladat: írjuk ki ugyanazt az adatot CSV-be és Parquet-be, majd ellenőrizzük a típusokat és a kerekítés nélküli kontrollösszeget. Bemenet a szállított lab/orders.csv, megfigyelési egysége egy rendelés, alapmérete 2400 sor. Kimenet: out/orders_roundtrip.csv, out/orders.parquet. Előzetes becslés: a sor- és oszlopszám nem változik, az egész minor unit összegek bitazonosak maradnak, a distance_km float64 összeg csak lebegőpontos tűréssel hasonlítható, a Parquet-fájl várhatóan kisebb lesz a CSV-nél, de a betöltött DataFrame RAM-igénye nem csökken a fájlmérettől.

8.1.2 Sor- és oszlopalapú tárolás

A CSV sororientált: a fájlban az 1. sor mezői, utána a 2. sor mezői következnek. Ha csak a distance_km oszlopra van szükség, akkor is be kell olvasni és értelmezni minden sort, mert a mezőhatárok csak így találhatók meg. Ez akkor kényelmes, ha egész sorokat fűzünk hozzá, például naplófájlnál.

A Parquet oszlopalapú: a fájl oszloponként tárolja az értékeket. A motor kihagyhatja a nem kért oszlopokat, és az egy oszlopon belüli értékek hasonlóságát is kihasználhatja tömörítésre. A gyakorlati következmény: analitikus lekérdezésnél, ahol kevés oszlop kell sok sorból, az oszlopos forma hatékonyabb; teljes sorok gyakori hozzáfűzésére viszont kevésbé természetes. A 7.4-ban említett zónatérkép is oszlopstatisztikára épül.

8.1.3 CSV idézés

A CSV-ben minden érték szöveg. A vessző elválasztó, ezért ha egy mező vesszőt, idézőjelet vagy újsort tartalmaz, idézőjelek közé kell tenni. Az RFC 4180 gyakorlat szerint az idézőjelen belüli idézőjel duplázással írható. Például:

order_id,note,amount
"A-1","contains ""quotes"" and, comma",123

A pandas read_csv alapértelmezése ezt várja: sep=",", quotechar='"', doublequote=True. A második mező értéke tehát contains "quotes" and, comma, és a sor három oszlopból áll. A CSV nem tárol típust; a 123 lehet int, string vagy akár dátum, ezt a betöltő dönti el. Ezért kell a sémát minden együttműködő félnek külön rögzítenie.

8.1.4 JSON normalizálás

A JSON gyakran beágyazott listákat és objektumokat hoz, például egy API-válasz rendelésenként items listát ad. Az elemzési tábla viszont lapos: egy sor egy megfigyelési egység. A pd.json_normalize a listát sorokká robbantja, a szülő mezőket pedig meta útvonallal ismétli.

import json
import pandas as pd

rows = json.loads("""[
  {"order_id": "A-1",
   "customer": {"customer_id": "C-9", "segment": "business"},
   "items": [{"line_id": 1, "qty": 2}, {"line_id": 2, "qty": 1}]}
]""")

items = pd.json_normalize(
    rows,
    record_path=["items"],
    meta=["order_id", ["customer", "customer_id"], ["customer", "segment"]],
)
print(items)

A kimeneti sorok az items elemei, a customer.customer_id és customer.segment oszlopok pedig a szülő objektumból ismétlődnek. A meta elemei útvonalak: ["customer", "customer_id"] a beágyazott kulcsot jelenti, nem két külön metaoszlopot.

8.1.5 Parquet séma és tömörítés

A Parquet-fájl a sémát is tárolja: minden oszlopnak fizikai típusa van, például int64, double vagy timestamp[us, tz=UTC], a string pedig UTF-8 bájttömb. Ezért a jól megírt Parquet-ből visszaolvasás után nem kell újra kitalálni, hogy a basket_minor egész szám-e. A pandas to_parquet a DataFrame dtype-jaiból PyArrow-táblát épít, és ezt a sémát a fájl lábjegyzetébe írja.

A tömörítés két rétegű. Először az oszlopon belüli kódolás dolgozik: a kevés egyedi értékű oszlop szótárat és indexeket kaphat, az ismétlődő értékek futamhossz-kódolást, az időbélyegek delta-kódolást. Erre jön az általános codec, alapértelmezetten snappy, opcióként zstd vagy gzip. A zstd általában kisebb fájlt ad, de több processzoridőbe kerül; a konkrét arányt csak az adott gépen mért futással szabad állítani.

Adatmennyiség elemi becsléseként: ha $n=2400$ sorban egy int64 oszlop van, a nyers bináris méret

$$S_{\text{int64}} = 8n = 19\,200\ \text{bájt}.$$

Ugyanez decimális szövegként $n \cdot \bar{\ell}$ bájt, ahol $\bar{\ell}$ az egy érték átlagos karakterhossza; az 1000 négy karakter, a 100000 hat. Egy $k$ egyedi értékű kategória szótárkódolt, durva alsó becslése

$$S_{\text{dict}} \approx k\bar{\ell}_v + n\left\lceil\frac{\log_2 k}{8}\right\rceil,$$

ahol $\bar{\ell}_v$ a szótárelemek átlagos hossza. A priority_flag esetén $k=2$, tehát egy bit is elég lenne soronként; a valóságban a tárolási overhead és a kódolási csoportosítás miatt ennél több lesz, de az elv világos: a típus és az ismétlődés is információ, amit a CSV szövegként eldob.

8.1.6 Partition pruning

A partíció a fájl elérési útjába emeli a gyakori szűrőt. Hive-stílusú könyvtár például:

out/orders_partitioned/year=2025/month=1/part-0.parquet
out/orders_partitioned/year=2025/month=2/part-0.parquet

A partíciós oszlop értéke a mappanévből olvasható, nem kell minden fájl tartalmát megnyitni. Ha a lekérdezés csak month = 2 sorokat kér, a motor pruninggal kihagyja a többi mappát. Ez különbözik a zónatérkép-pruningtól, amely egy oszlop min/max statisztikáját nézi a fájlon belül.

A partíció azonban nem automatikusan gyorsítás. Ha a kardináliság túl nagy, sok kis fájl keletkezik, és a sok metaadat-beolvasás meg a fájlmegnyitás többe kerül, mint maga a szűrés. A Northstar alapadata 180 napot fed le: napi partíció átlagosan 13 rendelést adna, ami jellemzően rossz aprózás; havi partíció 6 mappát ad, ami a havi szűréshez ésszerűbb.

8.1.7 Teljes minimális munkapélda

Mentsük a következő kódot s029_file_formats.py néven a lab projektgyökerébe, ahol az orders.csv található.

from pathlib import Path

import pandas as pd
import pyarrow.parquet as pq

INPUT = Path("orders.csv")
OUT = Path("out")
OUT.mkdir(exist_ok=True)

dtype = {
    "order_id": "string",
    "customer_id": "string",
    "basket_minor": "int64",
    "shipping_fee_minor": "int64",
    "quantity_sum": "int64",
    "item_rows": "int64",
    "distance_km": "float64",
    "priority_flag": "int64",
}
date_cols = ["order_ts", "actual_delivery_at", "label_available_at"]

orders = pd.read_csv(INPUT, dtype=dtype, parse_dates=date_cols)
for col in date_cols:
    orders[col] = pd.to_datetime(orders[col], utc=True)

n_rows, n_cols = orders.shape
basket_sum = int(orders["basket_minor"].sum())
shipping_sum = int(orders["shipping_fee_minor"].sum())
distance_sum = float(orders["distance_km"].sum())
n_unique_order = int(orders["order_id"].nunique())

csv_path = OUT / "orders_roundtrip.csv"
pq_path = OUT / "orders.parquet"
orders.to_csv(csv_path, index=False)
orders.to_parquet(pq_path, index=False)

csv_back = pd.read_csv(csv_path, dtype=dtype, parse_dates=date_cols)
for col in date_cols:
    csv_back[col] = pd.to_datetime(csv_back[col], utc=True)
pq_back = pd.read_parquet(pq_path)

checks = {
    "sorok": len(csv_back) == n_rows == len(pq_back),
    "oszlopok": list(csv_back.columns) == list(orders.columns) == list(pq_back.columns),
    "basket_minor sum": int(csv_back["basket_minor"].sum()) == basket_sum == int(pq_back["basket_minor"].sum()),
    "shipping_fee_minor sum": int(csv_back["shipping_fee_minor"].sum()) == shipping_sum == int(pq_back["shipping_fee_minor"].sum()),
    "distance_km sum": abs(float(csv_back["distance_km"].sum()) - distance_sum) < 1e-6
                     and abs(float(pq_back["distance_km"].sum()) - distance_sum) < 1e-6,
    "order_id uniq": int(pq_back["order_id"].nunique()) == n_unique_order == n_rows,
    "basket int64": str(pq_back["basket_minor"].dtype) == "int64",
}

print("Kontrollok:", checks)
print("CSV bájt:", csv_path.stat().st_size, "Parquet bájt:", pq_path.stat().st_size)
print("Parquet séma:")
print(pq.read_schema(pq_path))
print(orders.dtypes)

Futtassuk a projektgyökérből, bash vagy PowerShell munkakönyvtárból:

uv run python s029_file_formats.py

A futási kimenet a saját gépen mérhető; az alábbi csak szemléltető alak:

Kontrollok: {'sorok': True, 'oszlopok': True, 'basket_minor sum': True,
'shipping_fee_minor sum': True, 'distance_km sum': True, 'order_id uniq': True,
'basket int64': True}
CSV bájt: 567890 Parquet bájt: 123456
Parquet séma:
order_id: string
...

A konkrét bájtszámok és a séma pontos sorai a helyi orders.csv-ből és a telepített PyArrow-verzióból következnek; a checks mindegyikének viszont True-nak kell lennie.

8.1.8 Eredmény értelmezése

A kontrollösszeg a CSV-visszaolvasás után is egyezik, mert az egész minor unit oszlopokat int64-ként adtuk meg, és a pandas a float64 értékeket olyan szöveggé írta, amelyből ugyanaz a float64 áll vissza. A Parquet esetén a séma a fájlból jön: basket_minor fizikailag int64, distance_km double, a dátum timestamp[us, tz=UTC]. A distance_km összegnél a 1e-6 abszolút tűrés a lebegőpontos ábrázolás miatt indokolt; a sorrend azonos, ezért a hiba ennél jóval kisebb, de a lebegőpontos összeg nem egzakt egész.

A fájlméret-különbség a priority_flag, a kategóriák és az ismétlődő dátumrészek tömöríthetőségétől függ. A DataFrame RAM-igénye ettől független: a PyArrow visszaolvasás után a pandas ugyanúgy anyagi oszlopokat tart memóriában. A CSV-betöltésnél a dtype szótár nélkül a pandas találgathat; a Parquet-nél ez a találgatás megszűnik.

8.1.9 Feladatok

1. Értelmezés. Az alábbi CSV-sor hány mezőt tartalmaz, és mi a második mező pontos értéke?

"A-1","contains ""quotes"" and, comma",123

2. Implementálás. Írjuk ki out/orders_partitioned könyvtárba a rendeléseket year és month Hive-partícióval, majd PyArrow Dataset.scanner segítségével szűrjük ki a 2025. februári sorokat. Ellenőrző feltétel: a kiszűrt sorok száma egyezzen a month == 2 pandas-szűrés sorainak számával.

3. Hibadiagnózis. Az alábbi kód KeyError: 'customer_id' hibát ad. Mi a hiba oka, és mi a javítás?

rows = [{
    "order_id": "A-1",
    "customer": {"customer_id": "C-9", "segment": "business"},
    "items": [{"line_id": 1, "qty": 2}],
}]
pd.json_normalize(rows, record_path=["items"], meta=["customer_id"])

4. Megváltozott feltétel. A partner a distance_km mezőt szándékosan float32-re kerekítve kéri menteni, és zstd tömörítést használ. Ismételd meg a minimális példát ezzel a változtatással, és mérd a distance_km kontrollösszeg abszolút hibáját az eredeti float64 összeghez képest. Miért nem teljesülhet a 1e-6 tűrés?

8.1.10 Megoldások és indoklás

1. A sor három mezőből áll: order_id="A-1", note=contains "quotes" and, comma, amount=123. A dupla idézőjel a mező belsejében egyetlen idézőjelet jelent, a vessző a mezőn belül marad, mert a teljes mező idézőjelek között van. A pandas doublequote=True beállítása pontosan ezt a szabályt követi.

2. A minimális példa után futtatható kód:

import shutil

import pyarrow as pa
import pyarrow.dataset as ds

shutil.rmtree("out/orders_partitioned", ignore_errors=True)
orders_ym = orders.copy()
orders_ym["year"] = orders_ym["order_ts"].dt.year
orders_ym["month"] = orders_ym["order_ts"].dt.month

table = pa.Table.from_pandas(orders_ym, preserve_index=False)
ds.write_dataset(
    table,
    "out/orders_partitioned",
    format="parquet",
    partitioning=["year", "month"],
)

feb = ds.dataset(
    "out/orders_partitioned", format="parquet", partitioning="hive"
).scanner(filter=ds.field("month") == 2).to_table()

expected = int((orders_ym["month"] == 2).sum())
assert len(feb) == expected
print("február sorok:", len(feb), "hónapok:", sorted(feb["month"].unique().to_pylist()))

A pruningot a filter=ds.field("month") == 2 feltétel váltja ki: a motor a month=2 partíciós mappákat olvassa, a többit kihagyja.

3. A meta=["customer_id"] útvonal a legfelső szinten keresi a customer_id kulcsot, de az a customer objektumon belül van. A helyes útvonal ["customer", "customer_id"]. A javítás:

pd.json_normalize(
    rows,
    record_path=["items"],
    meta=["order_id", ["customer", "customer_id"], ["customer", "segment"]],
)

Így a normalizálás a customer_id szülőértéket minden tételsorhoz hozzárendeli.

4. Az átalakítás:

orders32 = orders.copy()
orders32["distance_km"] = orders32["distance_km"].astype("float32")
orders32.to_parquet("out/orders32.parquet", compression="zstd")
pq_back32 = pd.read_parquet("out/orders32.parquet")
error = abs(float(pq_back32["distance_km"].sum()) - distance_sum)
print("float32 abszolút hiba:", error)

A float32 értékenként körülbelül $2^{-24} \approx 6\cdot10^{-8}$ relatív kerekítést visz be, de 2400 érték összegénél az abszolút hiba az összeg nagyságrendjétől függ, és a tapasztalat szerint jóval meghaladja a 1e-6 tűrést. A zstd kisebb fájlt adhat a snappy alapértelmezésnél, de a kontrollösszegbeli különbséget a típuskerekítés okozza, nem a codec.

8.1.11 Gyakori tévedés ellenpéldával

Veszélyes azt hinni, hogy „minél több partíció, annál gyorsabb”. Ha a 2400 rendelést egyedi order_id-ként particionálnánk, körülbelül 2400 fájl keletkezne, átlagosan egy sorral. A pruning kiválasztana egyet, de a fájlrendszerben a sok kis fájl felsorolása, metaadat-olvasása és megnyitása drágább, mint a teljes 2400 sor beolvasása. Ugyanez a hiba kisebb formában a napi partíciónál: 180 nap, 13 sor fájlonként.

A másik gyakori félreértés: „a kisebb Parquet-fájl kevesebb RAM-ot igényel”. A tömörítés a tárolásra és az átvitelre hat. Betöltés után a DataFrame oszlopai memóriában anyagiasulnak: egy 2400 soros int64 oszlop 19 200 bájt, akár 2 KiB, akár 200 KiB volt a fájl. A fájlméret tervezési jel, a RAM-igény pedig a sorok számából és a fizikai típusokból számolható. Ez a 7.4-ban tanult vetítés és a 7.1-ben tanult sorjelentés közös következménye.

Önellenőrzés: Mit tárol a CSV a típusokból? Miért kell a meta útvonalat listaként megadni? Miért nem egyenlő a tömörített fájlméret a betöltött DataFrame memóriaigényével?

A Parquet oszlopos fájltárolása és az Arrow oszlopos memóriareprezentációja nem garantál teljes üzleti sémahelyességet. Az időzónás dátumok, decimális típusok és kategóriák oda-vissza alakítását külön tesztekkel érdemes ellenőrizni.

8.1.12 Rövid összefoglaló

A CSV szöveges és séma nélküli; az idézési szabályokat és a típusokat a betöltőnek külön kell ismernie. A JSON normalizálás a beágyazott listákból lapos táblát készít, a szülőmezőket meta útvonalakkal ismételve. A Parquet oszlopalapú, a fájlban tárolt sémával és kétrétegű tömörítéssel; az egész minor unit kontrollösszeg bitazonos, a float64 összegnél tűrés kell. A Hive-partíció pruninggal kihagyja a nem releváns mappákat, de a túlzott aprózás sok kis fájllal ront. A következő alfejezetben a DuckDB-t használjuk ugyanezen Parquet-állomány közvetlen SQL-lekérdezésére.