30.4 SQL-transzformáció és dbt helye

A feladat az elemzési transzformáció elválasztása a modellező Python-kódtól: a Northstar tranzakciós táblákból egy rendelésenként egy sort tartalmazó feature mart készül SQL-ben, és csak ez a mart megy a train/inference pipeline-ba. Bemenet a tiszta customers.csv (300 sor), orders.csv (2400 sor) és order_items.csv (4796 sor); a megfigyelési egység a martban egy rendelés. Cél: a mart sorszáma, tételaggregátumai és a szállítási díj összege egyezzen a forrástáblákból levezethető invariánsokkal. Előzetes becslés: a mart 2400 soros lesz, a customer-join egy-egy kapcsolat miatt nem változtatja a sorszámot, a tételszintű újraszámolt kosárérték rendelésenként egyezik a basket_minor értékkel. A dbt csak ezt a mart-definíciót teszi tesztelhetővé és újrafuttathatóvá; nem végez modellillesztést.

30.4.2 Fogalmi magyarázat: SQL-modell, source, test, materialization, lineage, adapter

SQL-modell (dbt model): dbt-ben a „modell” egy SELECT utasítás, amely egy táblát vagy nézetet definiál. Ez nem azonos a gépi tanulási modellel. A dbt a SELECT-ből generálja a CREATE TABLE AS vagy CREATE VIEW AS műveletet, és a modell neve lesz a kimeneti objektum neve. A modell hivatkozhat más modellekre a {{ ref('masik_modell') }} függvénnyel, nyers forrásokra pedig a {{ source('sema', 'tabla') }} függvénnyel.

Source: deklarált, külső tábla, amelyet nem a dbt hoz létre; a mi esetünkben a DuckDB-be már betöltött northstar.orders, northstar.customers és northstar.order_items. A source a nyers réteg, amelynek sémája a külső adatszerződésből származik. Sem a CSV-betöltés, sem a forrásrendszer elérése nem a dbt feladata; a dbt csak hivatkozik a meglévő táblákra.

Test: a dbt deklaratív ellenőrzéseket futtat. Generikus tesztek: unique (kulcs egyedi), not_null (kötelező mező), relationships (idegen kulcs a szülő táblába mutat), accepted_values (megengedett kategóriák). Egyedi tesztet is írhatunk: egy SQL lekérdezés, amely a hibás sorokat adja vissza; ha üres, a teszt sikeres. A mi kosáregyezési ellenőrzésünk ilyen egyedi teszt.

Materialization: a dbt kimeneti típusa. view (a SELECT-et minden lekérdezéskor futtatja), table (a SELECT eredményét fizikai táblába írja), incremental (csak az új részletet írja) és más adapter-specifikus változatok. A modellező Python-pipeline akkor kap stabil, ismételten olvasható bemenetet, ha a mart table materializationnel készül. A vékony staging-rétegeknél a view gyakran elég.

Lineage: a ref és source hívások függőségi éleket hoznak létre; ezekből irányított gráf áll össze. A dbt ezt használja a futtatási sorrendhez, és a dbt docs felületen megjeleníti. A lineage segít megválaszolni: ha a stg_order_items változik, mely mart-táblákat kell újraépíteni?

dbt adapter: a dbt magja és az adatbázis között álló komponens; a dbt-duckdb adapter például a DuckDB SQL-dialektusát és a DuckDB-hez illeszkedő materialization-t valósítja meg. A modellek nagyrészt hordozhatók, de a dátumfüggvények és a sémanyelvi részletek adapterfüggők lehetnek. A dbt-duckdb kombinációt ez a könyv keretjegyzete nem futtatta; a pontos telepítést és adapterverziót a hivatalos dbt-adapterdokumentáció alapján kell ellenőrizni.

30.4.3 Matematikai híd: adatmegmaradási invariánsok egy one-to-many join után

Jelölje egy adott i rendelés tételsorainak számát s_i (mértékegysége darab), a szállítási díját f_i (mértékegysége EUR-cent). Az order_items tábla összes sora

ahol N_orders = 2400 a rendelések száma. Ha a shipping_fee_minor oszlopot közvetlenül a tételszintű joinon keresztül összegezzük, az eredmény

ami nemnegatív díjak és legalább egy tétel rendelésenként esetén legalább akkora, mint a helyes rendelésszintű összeg

és szigorúan nagyobb, ha van olyan rendelés, amelyre s_i > 1 és f_i > 0. Hiányzó tételsor esetén belső csatlakoztatás a rendelést teljesen el is hagyhatja. Kézi számpélda: ha egy rendelésnél f_i = 500 cent és s_i = 3 tétel, a tételszintű join után 3 * 500 = 1500 cent jelenik meg az összegben, nem 500. Ugyanez a duplikáció érintené a target_late célváltozót is. A megoldás: a tételtáblát előbb rendelés szintre aggregáljuk, majd a rendelésszintű táblához joinoljuk. A három ellenőrizendő adatmegmaradási invariáns:

30.4.4 Teljes minimális munkapélda

Az alábbi Python-szkript a DuckDB Python API-val fut; a DuckDB az alap tanulási stack része. Feltételezzük, hogy a tiszta CSV-k a data/raw/ mappában vannak a lab/ munkakönyvtár alatt. A kimenet a mellékelt adatszerződésből levezetett elméletileg várt érték; e fejezet írásakor nem tartozik hozzá tényleges futási jegyzék.

# mart_build.py -- Linux/macOS, munkakönyvtár: lab/
# Szemléltető minimális munkapélda; a futási kimenet a helyi CSV-k tartalmától függ.
import duckdb

con = duckdb.connect()

con.execute("""
CREATE OR REPLACE TABLE orders AS
SELECT * FROM read_csv_auto('data/raw/orders.csv', header = true);

CREATE OR REPLACE TABLE customers AS
SELECT * FROM read_csv_auto('data/raw/customers.csv', header = true);

CREATE OR REPLACE TABLE order_items AS
SELECT * FROM read_csv_auto('data/raw/order_items.csv', header = true);

-- 1) Tételtáblából rendelésszintű aggregátum; nem joinolunk tételszinten.
CREATE OR REPLACE TABLE item_agg AS
SELECT
  order_id,
  SUM(quantity)                                AS agg_quantity,
  COUNT(*)                                     AS agg_item_rows,
  SUM(quantity * unit_price_minor - discount_minor)
                                               AS agg_basket_minor
FROM order_items
GROUP BY order_id;

-- 2) Rendelésenként egy sor: customer-join 1:1, item_agg-join 1:1.
CREATE OR REPLACE TABLE mart_order_features AS
SELECT
  o.order_id,
  o.customer_id,
  o.order_ts,
  o.promised_days,
  o.distance_km,
  o.priority_flag,
  o.basket_minor,
  o.shipping_fee_minor,
  o.target_late,             -- cél; a Python-pipeline explicit kiválogatja
  o.label_available_at,      -- split-ellenőrzéshez, nem prediktor
  ia.agg_quantity,
  ia.agg_item_rows,
  ia.agg_basket_minor,
  c.signup_date,
  c.segment,
  c.region,
  c.acquisition_channel,
  date_diff('day', c.signup_date, CAST(o.order_ts AS DATE))
                              AS account_age_days
FROM orders o
JOIN customers c
  ON o.customer_id = c.customer_id
LEFT JOIN item_agg ia
  ON o.order_id = ia.order_id;
""")

print("mart kontroll:")
print(con.execute("""
SELECT
  COUNT(*)                                              AS mart_count,
  COUNT(*) FILTER (WHERE agg_quantity IS NULL
                     OR agg_item_rows IS NULL
                     OR agg_basket_minor IS NULL)       AS missing_agg,
  COUNT(*) FILTER (WHERE agg_basket_minor <> basket_minor
                     OR agg_quantity <> quantity_sum
                     OR agg_item_rows <> item_rows)     AS mismatch_count,
  SUM(shipping_fee_minor)                               AS mart_shipping_sum
FROM mart_order_features;
""").fetchone())

print("source kontroll:")
print(con.execute("""
SELECT
  COUNT(*)                                               AS orders_count,
  SUM(shipping_fee_minor)                                AS orders_shipping_sum
FROM orders;
""").fetchone())

A mart_shipping_sum és az orders_shipping_sum értéke a generált CSV-k konkrét összege; a kontroll invariáns az egyenlőség, nem az előre rögzített szám. A mart shape-je (2400, 18): 18 oszlop, 2400 megfigyelés.

30.4.5 A dbt projektváza és a felelősségi térkép

A dbt-rész opcionális bővítő, nem a kezdőlabor telepített része. Az alábbi projektváz szemlélteti a struktúrát; a modellfájlok a fenti DuckDB SQL-lel logikailag azonos átalakítást írnak le. A dbt-duckdb adapter verzióját és a telepítési módot a hivatalos dokumentációban ellenőrizni kell; a parancsok Linux/macOS terminálban, lab/feature_mart munkakönyvtárban értendők, ha a projektet külön mappába másoljuk.

name: feature_mart
version: "1.0"
profile: feature_mart
model-paths: ["models"]
test-paths: ["tests"]

models:
  feature_mart:
    staging:
      +materialized: view
    mart:
      +materialized: table

SELECT
  order_id,
  customer_id,
  order_ts,
  promised_days,
  distance_km,
  priority_flag,
  basket_minor,
  shipping_fee_minor,
  quantity_sum,
  item_rows,
  target_late,
  label_available_at
FROM {{ source('northstar', 'orders') }}

SELECT
  order_id,
  line_id,
  product_id,
  quantity,
  unit_price_minor,
  discount_minor,
  quantity * unit_price_minor - discount_minor AS line_total_minor
FROM {{ source('northstar', 'order_items') }}

SELECT
  customer_id,
  signup_date,
  segment,
  region,
  acquisition_channel
FROM {{ source('northstar', 'customers') }}

WITH item_agg AS (
  SELECT
    order_id,
    SUM(quantity)                  AS agg_quantity,
    COUNT(*)                       AS agg_item_rows,
    SUM(line_total_minor)          AS agg_basket_minor
  FROM {{ ref('stg_order_items') }}
  GROUP BY order_id
)
SELECT
  o.order_id,
  o.customer_id,
  o.order_ts,
  o.promised_days,
  o.distance_km,
  o.priority_flag,
  o.basket_minor,
  o.shipping_fee_minor,
  o.target_late,
  o.label_available_at,
  ia.agg_quantity,
  ia.agg_item_rows,
  ia.agg_basket_minor,
  c.signup_date,
  c.segment,
  c.region,
  c.acquisition_channel,
  date_diff('day', c.signup_date, CAST(o.order_ts AS DATE)) AS account_age_days
FROM {{ ref('stg_orders') }} o
JOIN {{ ref('stg_customers') }} c
  ON o.customer_id = c.customer_id
LEFT JOIN item_agg ia
  ON o.order_id = ia.order_id

A date_diff DuckDB-függvény; más adapteren esetleg datediff(day, ...) kell. A models/mart/schema.yml a generikus tesztekkel:

version: 2

models:
  - name: mart_order_features
    description: "Rendelésenként egy sor; a Python train/inference pipeline ezt olvassa."
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: customer_id
        tests:
          - relationships:
              to: ref('stg_customers')
              field: customer_id
      - name: segment
        tests:
          - accepted_values:
              values: ['consumer', 'business']
      - name: region
        tests:
          - accepted_values:
              values: ['north', 'central', 'south']
      - name: priority_flag
        tests:
          - accepted_values:
              values: [0, 1]
      - name: basket_minor
        tests:
          - not_null

SELECT order_id
FROM {{ ref('mart_order_features') }}
WHERE COALESCE(agg_basket_minor, -1) <> basket_minor
   OR COALESCE(agg_quantity, -1)   <> quantity_sum
   OR COALESCE(agg_item_rows, -1)  <> item_rows

# Linux/macOS, munkakönyvtár: lab/feature_mart (opcionális projekt)
dbt run --select staging
dbt run --select mart
dbt test --select mart
dbt docs generate
dbt docs serve

A lineage gráf a ref/source élekből áll össze: stg_order_items → item_agg CTE a martban, stg_orders és stg_customers → mart_order_features. A dbt docs ezt jeleníti meg; az alábbi táblázat a felelősségi határokat rögzíti:

Réteg	Felelősség	Tipikus kimenet	Nem ez a dolga
Betöltő/ingest	CSV/forrás beolvasás, séma rögzítés	DuckDB nyers táblák	üzleti definíció
Staging	típus, tiszta elnevezés, source-teszt	`stg_*` view	aggregálás, üzleti logika
Mart	granularitás, join, feature-invariáns	`mart_order_features` table	tanítás, split, metrika
Python ML-pipeline	split, preprocessing, tanítás, kiértékelés	artifact, `metrics.json`, `split.json`	join-javítás, nyers tisztítás

30.4.6 Eredmény értelmezése

A mart kontrollja három dolgot ellenőriz: a 2400 soros szemcsézettséget, a tételszintű aggregátumok kosáregyezését és a szállítási díj nem többszöröződését. Ha a missing_agg vagy a mismatch_count 0, a mart rendelésszintű, és a tételtáblából származó oszlopok megegyeznek a rendeléstábla szerződött értékeivel. A mart_shipping_sum = orders_shipping_sum egyenlőség az invariáns; ha valaki tételszinten joinolt volna, a martösszeg a rendelésösszegnek a tételszámokkal súlyozott torzított változata lenne. A Python-modellező kód ezután nem végez join-javítást; a bemenet a mart szerződéséből jön. A target_late és label_available_at oszlopok a martban metaadatok; a train/inference kód felel azért, hogy a target_late-ot címkének, ne prediktornak használja. A séma-eltérést a modellező kód a Robusztusság és korlátokkal írt modellkártya szerinti bemeneti validációval külön ellenőrzi.

30.4.7 Feladatok

30.4.8 Megoldások és indoklás

1. A rossz mart minden rendeléshez annyi sort ad, ahány tételsora van: 3 sort. A shipping_fee_minor a joinban háromszorozódik, így az adott rendeléshez tartozó szállítási díjak összege 3 * 500 = 1500 cent, nem 500. A target_late = 1 szintén három sorban jelenik meg. A megsértett invariáns: |mart| = |orders| = 2400, és a szállításidíj-összegmegmaradás is sérül. A helyes eljárás a tételtábla előzetes GROUP BY order_id aggregálása.

-- models/staging/agg_order_items.sql
SELECT
  order_id,
  SUM(quantity)                               AS agg_quantity,
  COUNT(*)                                    AS agg_item_rows,
  SUM(quantity * unit_price_minor - discount_minor)
                                              AS agg_basket_minor
FROM {{ ref('stg_order_items') }}
GROUP BY order_id

version: 2

models:
  - name: agg_order_items
    description: "Rendelésenként egy sor: a tételszintű oszlopok aggregátumai."
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: agg_quantity
        tests:
          - not_null
      - name: agg_item_rows
        tests:
          - not_null
      - name: agg_basket_minor
        tests:
          - not_null

Ellenőrző invariant: a modell sorainak száma 2400, és a SUM(agg_basket_minor) egyezik a stg_orders.basket_minor oszlop összegével.

3. A tünet: COUNT(*) = 4796, miközben COUNT(DISTINCT order_id) = 2400; a mart soronként nem rendelés, hanem tétel. A szállítási díj összege nagyobb a rendeléstábla összegénél. Igazoló SELECT:

SELECT
  COUNT(*)                AS bad_mart_rows,
  COUNT(DISTINCT order_id) AS distinct_orders
FROM mart_order_features;

Elvárt diagnózis: bad_mart_rows = 4796, distinct_orders = 2400. A javítás:

WITH item_agg AS (
  SELECT order_id,
         SUM(quantity)     AS agg_quantity,
         COUNT(*)          AS agg_item_rows,
         SUM(line_total_minor) AS agg_basket_minor
  FROM {{ ref('stg_order_items') }}
  GROUP BY order_id
)
SELECT ...
FROM {{ ref('stg_orders') }} o
LEFT JOIN item_agg ia ON o.order_id = ia.order_id

és a unique teszt visszaállítása az order_id oszlopra. A zöld dbt test nem a modell helyességét bizonyította, hanem azt, hogy a definiált tesztek nem futottak a védendő invariánsra; ez a Kódminőség és együttműködő review alfejezetbeli review-elv: a tesztlefedettség a kódminőségi ellenőrzés része.

SELECT
  o.order_id,
  o.customer_id,
  o.order_ts,
  o.promised_days,
  o.distance_km,
  o.priority_flag,
  o.basket_minor,
  o.shipping_fee_minor,
  o.quantity_sum,
  o.item_rows,
  o.target_late,
  o.label_available_at,
  (SELECT COUNT(*)
   FROM {{ ref('stg_orders') }} prev
   WHERE prev.customer_id = o.customer_id
     AND prev.order_ts < o.order_ts
     AND prev.label_available_at < o.order_ts
  ) AS customer_order_count_before
FROM {{ ref('stg_orders') }} o

Fontos: a label_available_at < o.order_ts szűrés biztosítja, hogy a predikció pillanatában a korábbi rendelés kimenete már ténylegesen ismert; enélkül a jövőből szivárogna be címkeinformáció a feature-be. Ez a feltétel eltér a megszokott prev.order_ts < o.order_ts szűréstől, mert egy korábbi rendelés kézbesítése késhet. Egyedi tesztek:

-- tests/customer_order_count_before_is_nonnegative.sql
SELECT order_id
FROM {{ ref('stg_orders') }}
WHERE customer_order_count_before < 0;

-- tests/first_order_has_zero_history.sql
SELECT order_id
FROM {{ ref('stg_orders') }}
WHERE customer_order_count_before <> 0
  AND order_id IN (
    SELECT order_id
    FROM (
      SELECT order_id,
             ROW_NUMBER() OVER (
               PARTITION BY customer_id
               ORDER BY order_ts
             ) AS rn
      FROM {{ ref('stg_orders') }}
    ) ranked
    WHERE rn = 1
  );

A második teszt csak az ügyfél legkorábbi rendeléseit vizsgálja; ezeknél a feature kötelezően 0. A tesztelendő új oszlop dimenziója egész szám, tartománya nemnegatív.

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

A leggyakoribb tévedés: a dbt-t általános ML-optimalizálónak vagy adatbetöltőnek tekinteni. A dbt-ben a „model” SQL-transzformáció, nem illesztett gépi tanulási modell; nincs dbt run --select logistic_regression, mert a logisztikus regresszió a Python-pipeline artifactje, nem a dbt gráf csúcsa. A dbt nem gyűjt be minden adatforrást: a CSV-betöltés és a DuckDB-tábla létrehozása a source előtt külön ingest-réteg; a dbt seed csak kis referenciaadatokra való, nem a teljes tranzakciós adattóra. A dbt nem a randomizált experiment eszköze sem: a Northstar assignments/outcomes táblái külön kísérleti folyamatból jönnek, és nem az orders utólagos átalakításából, ahogyan a Kísérlet eredményének tudományos közlése is hangsúlyozza. Önellenőrzés: meg tudom-e magyarázni, miért marad 2400 sor a customer-join után? Meg tudom-e írni a kosáregyezési tesztet COALESCE-szel? Meg tudom-e nevezni azt a plusz feltételt, amely az előzményfeature-nél a jövőbeli címkeszivárgást akadályozza? Ha igen, a felelősségi határ világos.

30.4.10 Rövid összefoglaló

A dbt az elemzési SQL-transzformációt teszi tesztelhető, újrafuttatható és lineage-szerűen átlátható modellé. A source a már betöltött nyers táblára mutat; a staging nézetek tisztítanak; a mart tábla rendelésenként egy sorban, tételszintű aggregátummal és explicit célváltozóval készül. A generikus tesztek a kulcsot, idegen kulcsot és kategóriákat védik; egyedi tesztek a kosáregyezést és a szállításidíj-invariánst ellenőrzik. A join-kardinalitási hiba a tételszintű rendelésduplikációval mutatható ki. A következő alkalmazásban a Futásazonosító és kísérleti táblázat kísérletkövető táblázatnak a mart verzióját és a splitmanifestet is rögzítenie kell, mert a futásazonosító önmagában nem azonosítja a modell bemeneti feature-tábláját.

30.4.1 Cél és konkrét helyzet