Čištění, transformace a load dat

Cesta z bronze do silver vrstvy není jeden příkaz, ale sekvence kroků — a první z nich je profiling: zjistit, co v datech vlastně je, než začnete cokoliv měnit. Kolik je NULLů, jaké jsou rozsahy hodnot, kde jsou duplicity. Bez toho čistíte naslepo.

Nejrychlejší nástroje máte přímo v SQL a DataFrame API. ANALYZE TABLE spočítá statistiky (počty řádků, min/max, počty NULLů, distinct counts) — mimochodem stejné statistiky, které používá query optimizer pro plánování dotazů:

-- statistiky pro vsechny sloupce
ANALYZE TABLE sales.transactions COMPUTE STATISTICS FOR ALL COLUMNS;

-- detail jednoho sloupce: min, max, pocet NULL, distinct count
DESC EXTENDED sales.transactions amount;

df = spark.table("sales.transactions")

# count, mean, stddev, min, max pro numericke sloupce
df.describe().show()

# interaktivni profil vsech sloupcu vcetne histogramu
dbutils.data.summarize(df)

Utilita dbutils.data.summarize vykreslí interaktivní přehled distribucí, missing values a častých hodnot pro celý DataFrame — pro ad-hoc průzkum v notebooku je to nejrychlejší start.

Nad rámec příkazů nabízí Unity Catalog data quality monitoring se dvěma větvemi: anomaly detection (zapnete na úrovni schématu, automaticky hlídá freshness a completeness všech tabulek) a data profiling (hloubková per-table analýza distribucí, NULL procent a driftu, konfiguruje se pro jednotlivé tabulky přes záložku Quality v Catalog Exploreru nebo přes SDK). Profil generuje dvě metrické tabulky — profile metrics a drift metrics (KS test pro numerické, chi-kvadrát pro kategorické sloupce).

Na co se v profilu dívat: vysoká procenta NULL ve sloupcích, které mají být povinné; „unikátní" identifikátory, které unikátní nejsou; hodnoty mimo očekávané meze (záporné ceny); a změny distribucí v čase (data drift).

Špatný typ znamená ztrátu přesnosti, zbytečné úložiště nebo zablokované operace. Bronze vrstva často drží vše jako string — v silver je čas typy narovnat.

U celých čísel volte podle rozsahu hodnot: TINYINT (1 B, do 127) pro stavové kódy, INT (4 B) pro běžné identifikátory, BIGINT (8 B) pro velká ID. U desetinných čísel je rozhodnutí zásadnější:

U časových údajů rozhoduje, zda potřebujete časovou zónu: DATE pro kalendářní data bez času, TIMESTAMP pro události napříč zónami (interně normalizuje do UTC a při čtení aplikuje session timezone) a TIMESTAMP_NTZ, když má čas zůstat přesně tak, jak byl zadán (např. lokální čas schůzky). Kdo čeká, že TIMESTAMP vrátí přesně to, co uložil, může být překvapen — s jinou session timezone uvidí jinou hodnotu téhož okamžiku.

Pro vnořená data slouží komplexní typy: ARRAY<STRING> (seznam hodnot stejného typu — tagy produktu), MAP<STRING, STRING> (dynamické klíč–hodnota páry) a STRUCT (pevná sada pojmenovaných polí — adresa). Šetří joiny, ale komplikují některé operace; rozhodujte podle dotazovacích vzorů.

-- explicitni CAST hned pri loadu: chyby se projevi brzy, ne az v reportech
INSERT INTO silver.listings
SELECT
    CAST(raw_id AS INT) AS id,
    CAST(raw_price AS DECIMAL(12,2)) AS price,
    CAST(raw_listed_at AS TIMESTAMP) AS listed_at
FROM bronze.listings;

Duplicity vznikají opakovaným ingestem, integračními chybami nebo souběžnými zápisy. Nejdřív je najděte — klasicky přes GROUP BY ... HAVING COUNT(*) > 1, elegantněji přes QUALIFY s window funkcí, která vrátí rovnou celé duplicitní řádky:

-- najit duplicitni radky (vcetne vsech sloupcu)
SELECT * FROM sales.customers
QUALIFY COUNT(*) OVER (PARTITION BY customer_id, email) > 1;

-- ponechat jen nejnovejsi verzi kazdeho zakaznika
SELECT * FROM sales.customers
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id ORDER BY updated_at DESC
) = 1;

# odstraneni duplicit podle klicovych sloupcu (necha prvni vyskyt)
df_clean = df.dropDuplicates(["customer_id", "email"])

# uplna deduplicace pres vsechny sloupce
df_unique = df.distinct()

# zobrazit jen prebytecne duplicity: original minus deduplikovana verze
df_duplicates = df.exceptAll(df.dropDuplicates(["customer_id", "email"]))

NULL hodnoty reprezentují chybějící data a strategie závisí na kontextu: NULLy v povinných polích (identifikátory) → řádek zahodit; NULLy v numerických polích pro agregace → doplnit nulou či průměrem; NULLy v nepovinných kategoriích → placeholder „Unknown".

# zahodit radky s NULL v klicovych sloupcich
df_clean = df.dropna(subset=["customer_id", "amount"])

# doplnit ruzne defaulty pro ruzne sloupce
df_filled = df.fillna({
    "amount": 0,
    "status": "Unknown",
    "quantity": 1
})

V SQL zajistí totéž COALESCE(amount, 0). A jedna zrada navíc: v Pythonu nefiltrujte NULLy přes != None — NULL porovnání v SQL sémantice nevrací true/false, ale zase NULL. Vždy isNull() / isNotNull(), v SQL IS [NOT] NULL.

Filtrování, seskupování a agregace jsou základní trojice tvarování dat. V PySparku jsou filter() a where() identické metody; podmínky se skládají operátory & a | — každou podmínku do závorek:

from pyspark.sql.functions import col, count, sum, avg, max

# retezeni: filtr, seskupeni, agregace, razeni v jedne pipeline
df_result = (
    df_order
    .filter((col("o_orderstatus") == "F") & (col("o_totalprice") > 50000))
    .groupBy("o_orderpriority")
    .agg(
        count("o_orderkey").alias("n_orders"),
        sum("o_totalprice").alias("total_revenue"),
        avg("o_totalprice").alias("avg_order_value")
    )
    .sort(col("n_orders").desc())
)
display(df_result)

Díky lazy evaluation se celý řetězec vyhodnotí až při akci (display(), show()) a Spark si předtím celou sekvenci zoptimalizuje. SQL ekvivalent je klasika — WHERE pro řádky, GROUP BY pro skupiny, HAVING pro filtrování až po agregaci:

SELECT id, SUM(quantity) AS total_quantity
FROM dealer
GROUP BY id
HAVING SUM(quantity) > 20
ORDER BY total_quantity DESC;

Pravidlo z GROUP BY: každý sloupec v SELECT musí být buď v GROUP BY, nebo obalený agregační funkcí. Běžné agregace: count(), sum(), avg(), min(), max().

Dvě rodiny kombinování datasetů: joiny spojují tabulky horizontálně (přidávají sloupce podle klíče), set operátory vertikálně (skládají řádky dotazů se stejným schématem).

# left join; spolecny nazev klice jako string = bez duplicitniho sloupce
df_left = df_employee.join(df_department, on="deptno", how="left")

# anti join: zamestnanci bez existujiciho oddeleni (data quality check)
df_anti = df_employee.join(df_department, on="deptno", how="anti")

Set operátory: UNION defaultně odstraňuje duplicity, UNION ALL je nechává (a je rychlejší — přeskočí deduplikaci). INTERSECT vrací společné řádky, EXCEPT (synonymum MINUS) řádky prvního dotazu bez druhého.

Denormalizace zplošťuje normalizované tabulky do širších celků: dimenze se předjoinují do faktové tabulky, takže dotazy na ni už žádné joiny nepotřebují. Typicky se dělá při stavbě gold vrstvy a datových martů — cena je vyšší úložiště a nutnost refreshovat při změně zdrojů.

PIVOT otáčí unikátní hodnoty sloupce do samostatných sloupců („long" → „wide") — ideální pro křížové reporty. Vyžaduje agregační funkci, protože víc zdrojových řádků může mířit do téže buňky:

-- kvartaly z radku do sloupcu
SELECT year, region, Q1, Q2, Q3, Q4
FROM quarterly_sales
PIVOT (
  SUM(revenue)
  FOR quarter IN (1 AS Q1, 2 AS Q2, 3 AS Q3, 4 AS Q4)
);

Výsledek: každý region má kvartální tržby vedle sebe v jednom řádku. V jednom pivotu lze použít i víc agregací najednou (SUM(revenue) AS total, AVG(revenue) AS avg) — vzniknou sloupce Q1_total, Q1_avg atd.

UNPIVOT je opačný směr („wide" → „long"): měsíční sloupce se přetočí do párů název–hodnota, což je formát vhodný pro agregace a ML pipelines:

SELECT * FROM regional_targets
UNPIVOT (
  target_amount FOR month IN (jan, feb, mar, apr)
);

Z každého měsíčního sloupce vznikne řádek s hodnotou target_amount a názvem měsíce ve sloupci month. NULL hodnoty se defaultně vynechávají — pokud je chcete, přidejte UNPIVOT INCLUDE NULLS. Sloupcům lze dát aliasy (jan AS `January`).

Poslední krok pipeline: dostat vyčištěná data do cílové tabulky. Volba strategie závisí na vztahu nových dat k existujícím:

MERGE INTO je nejuniverzálnější: spáruje zdrojové řádky s cílovými podle klíče v ON a pro každou větev (matched / not matched) definuje akci — proto se mu říká upsert:

MERGE INTO customers AS target
USING customer_updates AS source
ON target.customer_id = source.customer_id
WHEN MATCHED AND source.price <> target.price THEN
    UPDATE SET target.price = source.price,
               target.updated_at = current_timestamp()
WHEN MATCHED AND source.discontinued = true THEN
    DELETE
WHEN NOT MATCHED THEN
    INSERT *;

V PySparku poskytuje totéž Delta Lake API — DeltaTable.forName(...).merge(...) s metodami whenMatchedUpdateAll() a whenNotMatchedInsertAll() pro případ shodných schémat.

Realitní kancelář Pristine Properties působící ve velkých nizozemských městech nasbírala inzeráty nemovitostí z poboček a satelitních databází — s nepřesnými cenami, duplicitními inzeráty, NULLy v klíčových polích a statistikami trhu v širokém formátu. Vaším úkolem je data zprofilovat, narovnat datové typy cen a timestampů, vyřešit duplicity a chybějící hodnoty v PySparku, spojit inzeráty s makléři a prodeji přes inner/left joiny a přetvarovat statistiky pomocí SQL PIVOT/UNPIVOT.

▶ Otevřít lab na GitHubu (EN) · Learn modul k tématu — s krátkými výukovými videy