</> Delta Table Crash Course

A architektúra-térkép azt mutatja meg, hogyan áll össze egy tipikus data engineering platform a forrásrendszerektől a fogyasztásig. A hatszakaszos folyamat (forrás, beérkeztetés, tárolás, feldolgozás, metaadat, fogyasztás) szinte minden modern adatplatformon megtalálható valamilyen formában, függetlenül attól, hogy Databricks, Snowflake, vagy saját építésű megoldásról van szó.

A diagram vizuálisan is megmutatja az adatáramlás irányát: balról jobbra halad a nyers adat a forrásrendszerektől a feldolgozott, riportkész adatokig. Minden szakasznak megvannak a saját tipikus eszközei és technológiái, de a Delta Lake a "tárolás" sávban tölt be kulcsszerepet, mint a nyílt táblázatformátum, amely összeköti a feldolgozó rendszereket a tároló réteggel.

Ez a notebook teljesen bootstrapelt: a projekt gyökerében lévő bootstrap.ps1 vagy bootstrap.sh létrehozza a .venv virtuális környezetet, telepíti a requirements.txt csomagjait, és regisztrálja a Jupyter kernelt. Ez biztosítja, hogy mindenki ugyanabban a környezetben futtatja a kódot, függetlenül attól, hogy milyen operációs rendszert használ.

A data engineering projektekben a reproduckálható környezet kiemelkedően fontos: ha a csapattagok különböző csomagverziókkal dolgoznak, akkor könnyen előfordulhat, hogy az egyik gépen működő pipeline a másikon hibát dob. A virtuális környezet (venv) ezt a kockázatot szünteti meg úgy, hogy minden függőséget a projekt könyvtárában tart, a globális Python telepítés érintetlenül hagyása mellett.

A kódban a REQUIRED_IMPORTS lista határozza meg, mely csomagoknak feltétlenül elérhetőnek kell lenniük a futás elkezdése előtt. Az os.name ágon a szkript megkülönbözteti a Windows és Linux/macOS környezeteket, így a megfelelő bootstrap szkriptet hívja meg. A subprocess.check_call() biztosítja, hogy ha a bootstrap bármely lépése sikertelen, a notebook azonnal jelezze a hibát.

Tip: Ha a bootstrap lefutott, a .venv mappa tartalmazza a teljes Python környezetet -- ezt érdemes hozzáadni a .gitignore-hoz, mert a gépen generálódik, nem pedig verziózandó.

A példához a deltalake Python csomagot használjuk. Ez a delta-rs projekt Python bindingja, és Spark nélkül is tud Delta táblát írni/olvasni. A delta-rs egy Rust nyelven írt, nagy teljesítményű Delta Lake implementáció, amely lehetővé teszi, hogy nem csak JVM-alapú rendszerekben (pl. Spark) használjuk a Delta formátumot, hanem tiszta Python környezetben is.

A data engineering gyakorlatban ez azért jelentős, mert sok ETL pipeline nem igényel elosztott feldolgozást -- egy egyszerű Python scripttel is létrehozhatsz production-grade Delta táblákat, amelyek aztán bármely Delta-kompatibilis eszközzel (Spark, Databricks, Trino, Presto) olvashatók. A pyarrow biztosítja a hatékony memórián belüli adatkezelést, a pandas pedig a kényelmes DataFrame API-t.

A kódcella meghívja a sys.executable útvonalat, valamint a három fő csomag verziószámát, hogy ellenőrizhesd: a megfelelő virtuális környezet aktív, és a verziók kompatibilisek egymással.

Tip: A deltalake csomag verziója határozza meg, mely Delta protokollverziókat tudja írni és olvasni -- régebbi verziók nem támogatják például a column mapping vagy a deletion vectors funkciókat.

A Delta table fizikailag egy mappa lesz a gépeden. Ebben lesznek a Parquet adatfájlok és a _delta_log könyvtár. Ez a kettősség a Delta Lake alapja: az adatok maguk Parquet formátumban tárolódnak (amely már önmagában is oszlopos és tömörített), de a tranzakciós napló adja hozzá az ACID garanciákat, a verziózást és a sémakezelést.

A kódban a base_dir egy gyűjtőmappa, amely alá az összes demo Delta tábla kerülni fog. A shutil.rmtree() hívás biztosítja, hogy minden futtatásnál tiszta lappal induljunk -- ez a notebook kontextusában praktikus, de production környezetben természetesen nem törölnénk az adatokat.

A DeltaTable import a meglévő Delta táblák olvasásához kell, a write_deltalake pedig az íráshoz. A Path osztály a pathlib-ből platformfüggetlen útvonalkezelést biztosít, így Windows és Linux alatt is ugyanaz a kód működik.

Tip: A base_dir.mkdir(parents=True, exist_ok=True) létrehozza a szülőkönyvtárakat is, ha még nem léteznek -- hasznos, ha mélyebb könyvtárszerkezetet szeretnél felépíteni.

Ez lehetne CSV-ből, API-ból vagy adatbázisból jövő adat is. Most kézzel készítünk egy kis táblát, amely rendelési adatokat tartalmaz. A DataFrame négy rendelést mutat be, mindegyikhez rendelésszám (order_id), vásárló azonosító (customer_id), összeg (amount), dátum (order_date) és forrásrendszer (source_system) tartozik.

A valós adatfolyamatokban az ilyen forrásadat általában egy OLTP adatbázisból, egy REST API válaszából vagy egy beérkező CSV fájlból származik. A pandas DataFrame egy univerzális közbenső formátum: szinte bármilyen adatforrást be tudunk tölteni bele, majd onnan a write_deltalake segítségével Delta táblaként kiírni. A pd.to_datetime() biztosítja, hogy a dátum oszlop valóban dátumtípusként legyen tárolva, nem pedig szövegként.

A source_system oszlop jelzi, hogy a rendelés honnan érkezett -- ez később a bronze/silver rétegekben fontos lesz, amikor forrásalapú szűréseket vagy statisztikákat készítünk.

Tip: Éles adatfolyamatokban a forrásadatot soha nem módosítjuk azonnal -- először a bronze rétegbe mentjük eredeti formájában, hogy vissza lehessen nyomozni bármilyen adatminőségi problémát.

Itt történik a lényeg. A write_deltalake(...) létrehozza a Delta table-t a megadott mappában. Ez az a pont, amikor egy egyszerű pandas DataFrame-ből egy verziózott, ACID tranzakciókat támogató táblává válik az adat. A függvény a háttérben Parquet fájlokat hoz létre az adatoknak, és egy _delta_log könyvtárat a metaadatoknak.

Nem kell hozzá: Databricks, Synapse, Azure, metastore, cluster. Ez az egyik legfontosabb tanulság: a Delta Lake egy nyílt formátum, amelyhez csak egy kompatibilis writer kell. A deltalake Python csomag éppen ilyen writer, és bármilyen fájlrendszerre tud írni -- legyen az lokális mappa, S3 bucket vagy ADLS container.

A write_deltalake() első paramétere a célmappa útvonala stringként, a második pedig az adat (pandas DataFrame, PyArrow Table vagy RecordBatchReader). Alapértelmezetten mode="overwrite" módban fut, ami azt jelenti, hogy ha már létezik tábla a mappában, felülírja. Később látni fogjuk az append módot is.

Tip: Ha a célmappa már tartalmaz egy Delta táblát és más sémával írsz rá, a write_deltalake alapértelmezetten hibát dob -- ez véletlen sématörés ellen véd. Ha szándékosan módosítani akarod a sémát, a schema_mode="overwrite" paramétert kell használnod.

A Delta table mappájában Parquet fájlokat és egy _delta_log mappát kell látnod. A kódcella a rglob("*") segítségével rekurzívan listázza ki a tartalmat, és megjelöli, hogy mi könyvtár ([DIR]) és mi fájl ([FILE]).

A Parquet fájl tartalmazza magát az adatot -- ebben az esetben a négy rendelési sort oszlopos, Snappy tömörítéssel. A fájl neve tartalmaz egy egyedi azonosítót (UUID), ami garantálja, hogy akkor sem lesz névütközés, ha több writer párhuzamosan ír ugyanabba a táblába. A _delta_log könyvtárban pedig a tranzakciós napló található, amely a következő sectionben részletesen bemutatásra kerül.

Fontos megérteni, hogy ez a kettősség (adatfájlok + napló) az, ami megkülönbözteti a Delta táblát egy sima Parquet fájlokra írt mappától. Ha csak Parquet fájlok lennének, nem lenne sémakezelés, nem lenne verziózás, és nem lennének ACID tranzakciók sem.

Tip: Soha ne manuálisan módosítsd vagy töröld a Parquet fájlokat a Delta tábla mappájában -- mindig a Delta API-n keresztül kezeldd az adatokat, különben a tranzakciós napló és az adatfájlok inkonzisztenssé válnak.

A _delta_log JSON fájlok írják le a tábla tranzakcióit. Itt van például a schema és az, hogy mely Parquet fájlok tartoznak a táblához. Minden Delta log fájl egy-egy verziót (commitot) képvisel: a 00000000000000000000.json a 0. verzió, a 00000000000000000001.json az 1. verzió, és így tovább.

A log fájl nem egyetlen JSON objektum, hanem JSON Lines formátumú: minden sor egy önálló JSON objektum, egy-egy "action"-t ír le. Az első kódcella kilistázza a log fájlokat, a második pedig beolvassa és formázottan kiírja az első log tartalmát. A json.loads() hívás soronként dolgozza fel a fájlt, a json.dumps(..., indent=2) pedig ember által olvashatóra formázza a kimenetet.

A log tartalmában négyféle bejegyzést fogsz látni: protocol (Delta protokoll verzió), metaData (tábla azonosító, séma, formátum), add (hozzáadott Parquet fájl elérési útja és mérete), valamint commitInfo (milyen művelet történt, pl. CREATE_TABLE). Ez a felépítés teszi lehetővé a time travel-t és az ACID garanciákat.

Tip: A Delta log fájlok kis méretűek, de nagyon sok apró fájl keletkezhet sok írás esetén -- éles rendszerekben az OPTIMIZE és a log compaction segít ezt kezelni.

Most nincs metastore. Nem tudjuk azt mondani, hogy SELECT * FROM silver.sales, mert nincs ilyen regisztrált táblanév. De path alapján tökéletesen olvasható a Delta table. A DeltaTable(path) konstruktor megnyitja a táblát, a .to_pandas() pedig az egész tábla tartalmát egy pandas DataFrame-ként adja vissza.

A path-alapú olvasás a legegyszerűbb módja a Delta táblák elérésének, és teljesen elegendő ETL jobok, ad hoc elemzések és notebookok esetén. A DeltaTable objektum nem tölti be azonnal az összes adatot a memóriába -- csak a metaadatokat olvassa be, és az adatok csak a .to_pandas() vagy .to_pyarrow() hívásakor kerülnek betöltésre.

Ha lenne metastore (pl. Databricks Unity Catalog vagy Hive Metastore), akkor a táblanév alapján is elérnéd ugyanezt az adatot, mert a metastore leképezi a logikai nevet a fizikai útvonalra. De a tábla önmagában, a fizikai fájlokkal, metastore nélkül is teljes értékűen létezik és olvasható.

Tip: Nagy táblák esetén a .to_pyarrow() jobb választás lehet a .to_pandas() helyett, mert a PyArrow kevesebb memóriát használ és lazy evaluációt támogat.

A schema nem a storage accountból jön. A schema a Delta metadata része lett, amikor megírtuk a táblát. A dt.schema().json() hívás a tranzakciós napló metaData bejegyzéséből olvassa ki a séma információt, nem pedig magukból a Parquet fájlokból (bár azok is tartalmaznak sémát).

Ez azért fontos, mert így a Delta Lake garantálja a séma konzisztenciáját: mindenki, aki olvassa a táblát, ugyanazt a sémát kapja, függetlenül attól, hogy melyik eszközt használja. A séma tartalmazza az oszlopok nevét, típusát, nullability-jét és egyéb metaadatokat. A timestamp típus például mikrosecundumos pontossággal és UTC időzónával van tárolva, ahogy az a kimenetben is látszik.

A kimenetből jól látható, hogy a order_id integer, a customer_id string, a order_date pedig egy strukturált timestamp típus. Ez a típusinformáció a Delta logban van rögzítve, és minden új adatíráskor validálva lesz a séma ellen.

Tip: A Delta Lake támogatja a séma evolúciót is (schema_mode="merge"), ami új oszlopok hozzáadását teszi lehetővé anélkül, hogy a meglévő adatokat újra kellene írni.

Most appendelünk két új sort. Ettől új Parquet fájl és új Delta log verzió jön létre. A write_deltalake(..., mode="append") nem írja felül a meglévő adatokat, hanem hozzáadja az új adatokat a táblához, létrehozva egy új verziót a tranzakciós naplóban.

A kódban először létrehozunk egy új DataFrame-et (new_sales_df) két rendeléssel, majd az append módú írással hozzáadjuk a meglévő táblához. Utána újra megnyitjuk a DeltaTable objektumot, hogy lássuk a frissített tartalmat. Az eredménytáblában már 6 sort fogsz látni -- az eredeti 4-et és az újonnan hozzáadott 2-t, az utóbbiakat a kimenetben zöld színnel kiemelve.

Az append művelet a Delta Lake-ben atomi (atomic): vagy teljes egészében sikeres, vagy egyáltalán nem történik meg. Ez az ACID "A" (Atomicity) garancia, ami azt jelenti, hogy soha nem fogod látni félig írt adatot, ami production rendszerekben kritikus fontosságú.

Tip: A DeltaTable(path) objektumot újra kell hozzárendelni az append után, mert a régi objektum még a régi verzióra hivatkozik -- a Delta tábla nem dinamikusan frissíti magát a memóriában.

A Delta log miatt a tábla verziózott. Az első írás volt a 0. verzió, az append után létrejött az 1. verzió. A kód első cellája kilistázza a _delta_log mappában lévő JSON fájlokat, a második cella pedig bemutatja a time travel funkciót: a DeltaTable(path, version=N) paraméterrel bármelyik korábbi verziót be tudod tölteni.

A time travel az egyik legerősebb Delta Lake funkció, mert lehetővé teszi az adatok bármely korábbi állapotának lekérdezését. Ez auditálásra, hibakeresésre, ML kísérletek reprodukálására és adatvisszaállításra is használható. A 0. verzió 4 sort tartalmaz (az eredeti írás), az 1. verzió pedig már 6-ot (az append után).

A színfalak mögött a Delta Lake úgy oldja meg a time travel-t, hogy minden verzió log fájlja feljegyzi, mely Parquet fájlok tartoznak az adott verzióhoz. Amikor egy régi verziót kérsz, a reader összeállítja a megfelelő fájllistát a log fájlokból, és csak azokat olvassa be. A régi adatfájlok mindaddig megmaradnak, amíg a VACUUM parancs törli őket.

Tip: Alapértelmezetten a VACUUM 7 napnál régebbi fájlokat töröl -- ha hosszabb time travel ablakra van szükséged, állítsd be a delta.logRetentionDuration táblatulajdonságot.

Az ETL és az ELT nem fájlformátum, nem Databricks feature. Inkább arról szól, hogy mikor történik a transzformáció. Az ETL (Extract-Transform-Load) esetén az adatot már a betöltés előtt megtisztítod, típusozod és üzleti logikával látod el, és csak a kész eredményt töltöd be a célrendszerbe. Az ELT (Extract-Load-Transform) esetén viszont a nyers adatot gyorsan betöltöd a célrendszerbe, és csak utána transzformálod.

ETL = Extract → Transform → Load
ELT = Extract → Load → Transform

A lakehouse architektúrában az ELT a domináns megközelítés, mert a nyers adat megőrzése (bronze réteg) biztosítja az újrafeldolgozhatóságot, az auditálhatóságot és a rugalmasságot. Ha az üzleti logika változik, egyszerűen újrafuttathatod a bronze → silver transzformációt, anélkül hogy vissza kellene menned a forrásrendszerhez. A táblázat összehasonlítja a két megközelítés jellemzőit.

ETL esetén először Pythonban megtisztítjuk és üzleti oszlopokat képzünk, és csak a kész adatot írjuk ki. Ez a klasszikus ETL (Extract-Transform-Load) minta: az adat már a Delta táblába íráskor tisztított, típusozott és üzletileg felhasználható formátumú.

A demo három cellából áll: az első létrehozza a nyers forrásadatot (ahogy egy OLTP rendszerből érkezhetne, szöveges mezőkkel, formázatlan összegekkel), a második elvégzi a transzformációt (szóközök eltávolítása, típuskonverzió, nettó összeg számítása, logikai oszlop képzése), a harmadik pedig kiírja az eredményt egy Delta táblába.

A transzformációs lépésben a str.replace(" ", "") eltávolítja a számokból az ezres elválasztó szóközöket, az astype("int64") számmá konvertál, a nettó összeg képzése pedig a bruttó összeg és a kedvezmény százalék alapján történik. Az is_online oszlop egy logikai jelző, amely egyszerűbb szűréseket tesz lehetővé.

Tip: Az ETL minta hátránya, hogy ha a transzformációs logika változik, a teljes folyamatot újra kell futtatni -- a nyers adat már nincs sehol tárolva ebben a megközelítésben.

Batch feldolgozásnál egyszerre egy véges adatcsomagot dolgozol fel. Például: minden éjjel 01:00-kor beolvasod az előző napi CSV-ket, frissíted a bronze, silver és gold rétegeket. Ez a legelterjedtebb feldolgozási mód a data engineeringben, mert egyszerűen megtervezhető, könnyen újrafuttatható, és a hibakeresés is viszonylag egyértelmű.

A kódcella egy összefoglaló táblázatban mutatja be a batch pipeline lépéseit: az extract fázisban CSV fájlokat olvasunk be, a load bronze-ba írjuk a nyers adatot, a transform silver-ben tisztítjuk, végül az aggregate gold-ban aggregáljuk a KPI-kat. Minden lépés batch jellegű, azaz egyszerre dolgozza fel a teljes adatcsomagot.

A batch feldolgozás fő előnye az egyszerűség: ha valami hiba történik, újra lehet futtatni az egész folyamatot (idempotens pipeline), és a végeredmény garantáltan konzisztens lesz. A hátránya, hogy az adat frissessége a batch ütemezésétől függ -- ha naponta fut, akkor a riportok legfeljebb egy napos adatot mutatnak.

Tip: A batch pipelineok tervezésénél érdemes az idempotenciára törekedni: azaz a többszori futtatás ugyanazt az eredményt adja, nem duplikál adatot, és nem okoz mellékhatást.

Valódi streaminghez Spark Structured Streaming, Flink, Kafka kell. Itt micro-batch szimulációt csinálunk, amely bemutatja a streaming gondolkodásmód alapjait anélkül, hogy valódi streaming infrastruktúrára lenne szükség. A micro-batch lényegében kicsi, gyakran ismétlődő batch-ek sorozata -- a legtöbb "streaming" rendszer (beleértve a Spark Structured Streaming-et is) valójában így működik a színfalak mögött.

A kódcella két batch-et szimulál, mintha azok időben egymás után érkeznének (például 30 másodpercenként). Minden batch esetén a teljes pipeline lefut: a nyers adat a bronze táblába kerül (az első batch overwrite, a többi append), majd a bronze-ból újraolvassuk az összes eddigi adatot, transzformáljuk silver-re, aggregáljuk gold-ra, és felülírjuk a silver/gold táblákat. Így minden batch után a silver és gold táblák a teljes eddigi történetet tükrözik.

A különbség az egyszerű batch és a micro-batch között az, hogy itt az adat folyamatosan érkezik és feldolgozásra kerül, nem pedig egyetlen nagy futásban. A _batch_id oszlop a bronze rétegben nyomon követi, melyik adat melyik batch-ből származik, ami segíti az auditálást és a hibakeresést.

Tip: Éles streaming rendszerekben a checkpointing és a watermarks biztosítják, hogy a rendszer összeomlás esetén pontosan onnan folytatódhassa a feldolgozást, ahol abbamaradt -- ez a szimulációban nem jelenik meg, de production-ben elengedhetetlen.