Data quality constraints

Nekvalitní data se chovají jako kontaminace — jakmile projdou do silver a gold vrstvy, šíří se do reportů, ML modelů a rozhodnutí. Dokumentace to formuluje přímo: nevalidní data „can derail analytics projects, corrupt business reports, and erode confidence in your data platform". Proto se kvalita dat nehlídá na jednom místě, ale ve vrstvách — každá vrstva chytá jiný typ problému a co proklouzne jednou, zachytí další.

Azure Databricks nabízí čtyři vestavěné mechanismy: schema enforcement (Delta Lake automaticky validuje strukturu a typy při zápisu), table constraints (pravidla přímo na tabulce, odmítnou zápis nevalidního řádku), pipeline expectations (deklarativní pravidla v Lakeflow pipelines s volitelnou reakcí) a monitoring (metriky kvality v UI a event logu). Kombinace všech čtyř se v dokumentaci označuje jako „defense in depth".

Nejjednodušší validace je NOT NULL constraint — deklaruje se přímo v definici sloupce nebo se doplní později přes ALTER TABLE. Delta Lake pak odmítne každou transakci, která by do sloupce zapsala NULL.

-- NOT NULL primo v definici tabulky
CREATE TABLE main.silver.customers (
    customer_id INT NOT NULL,
    email STRING NOT NULL,
    first_name STRING,
    last_name STRING
);

-- dodatecne pridani na existujici tabulku
ALTER TABLE main.silver.customers ALTER COLUMN email SET NOT NULL;

Složitější byznys pravidla — rozsahy hodnot, formáty, kombinace sloupců — pokrývá CHECK constraint. Je to libovolný booleovský SQL výraz vyhodnocovaný při každém INSERT i UPDATE; porušení znamená selhání celé transakce.

ALTER TABLE main.silver.employees
  ADD CONSTRAINT valid_age CHECK (age >= 18 AND age <= 120);

ALTER TABLE main.silver.transactions
  ADD CONSTRAINT positive_amount CHECK (amount > 0);

-- PK je jen metadata, unikatnost NEvynucuje
CREATE TABLE main.silver.orders (
    order_id INT NOT NULL,
    customer_id INT,
    order_date DATE,
    CONSTRAINT orders_pk PRIMARY KEY (order_id)
);

-- realnou kontrolu duplicit resi napr. materialized view
CREATE OR REFRESH MATERIALIZED VIEW order_validation AS
SELECT order_id, COUNT(*) AS occurrence_count
FROM orders
GROUP BY order_id
HAVING COUNT(*) > 1;

Řetězec tam, kde má být číslo, rozbije agregace a výpočty. První obrannou linií je opět schema enforcement: Delta Lake při zápisu porovnává typy příchozích sloupců se schématem cílové tabulky. Dokumentace: „All inserted columns must exist in the target table" a typy musí sedět nebo být bezpečně převoditelné. Zápis hodnoty 'fifty' do INT sloupce tedy selže sám od sebe.

Jenže bronze vrstva bývá plná řetězců (CSV, JSON) a převod na správné typy je vaše práce. K tomu slouží dvojice funkcí s zásadně odlišným chováním:

-- try_cast: selhani prevodu = NULL, ktery umim detekovat
SELECT
    raw_amount,
    try_cast(raw_amount AS DECIMAL(10,2)) AS validated_amount,
    CASE WHEN try_cast(raw_amount AS DECIMAL(10,2)) IS NULL
         THEN 'Invalid amount format' ELSE 'Valid' END AS validation_status
FROM staging_data;

Na tomto vzoru stojí praktický karanténní pattern: validní řádky přetypované do silver tabulky, nevalidní do karanténní tabulky s časovou značkou a důvodem. Nic se neztratí a datový tým má co reklamovat u zdroje.

-- validni zaznamy do silver
INSERT INTO silver_transactions
SELECT transaction_id,
       cast(amount AS DECIMAL(10,2)) AS amount,
       cast(transaction_date AS DATE) AS transaction_date
FROM bronze_transactions
WHERE try_cast(amount AS DECIMAL(10,2)) IS NOT NULL
  AND try_cast(transaction_date AS DATE) IS NOT NULL;

-- nevalidni do karanteny s duvodem
INSERT INTO quarantine_transactions
SELECT transaction_id, amount AS raw_amount, transaction_date AS raw_date,
       current_timestamp() AS quarantined_at,
       'Type validation failed' AS reason
FROM bronze_transactions
WHERE try_cast(amount AS DECIMAL(10,2)) IS NULL
   OR try_cast(transaction_date AS DATE) IS NULL;

Schema drift je jiný problém než špatné hodnoty: mění se samotná struktura dat. Zdrojový systém přidá sloupec, přejmenuje pole, změní typ — a pipeline, která to nečeká, buď spadne, nebo hůř: data potichu ztratí. Výchozí chování Delta Lake je konzervativní: zápis s neznámými sloupci selže. To je bezpečné, ale u živých zdrojů potřebujete strategii.

Auto Loader nabízí pět režimů přes volbu cloudFiles.schemaEvolutionMode:

Nejužitečnější kompromis pro produkci je režim rescue: pipeline nikdy nespadne kvůli struktuře a nic se neztratí. Vše, co neodpovídá očekávanému schématu, přistane jako JSON ve sloupci _rescued_data, který můžete periodicky vyhodnocovat.

(spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/Volumes/main/raw/schema")
    .option("cloudFiles.schemaEvolutionMode", "rescue")
    .option("rescuedDataColumn", "_rescued_data")
    .load("/Volumes/main/raw/claims")
    .writeStream
    .option("checkpointLocation", "/Volumes/main/raw/checkpoint")
    .toTable("main.bronze.claims")
)

Pro batch zápisy a MERGE existují protějšky: .option("mergeSchema", "true") u zápisu DataFramu a syntaxe MERGE WITH SCHEMA EVOLUTION INTO v SQL — obě automaticky přidají nové sloupce do cílové tabulky.

MERGE WITH SCHEMA EVOLUTION INTO target_table t
USING source_data s
ON t.customer_id = s.customer_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

Expectations v deklarativních pipelines jsou pravidla kvality, která se vyhodnocují nad každým záznamem procházejícím pipeline. Každá expectation má tři části: jméno (identifikuje pravidlo v monitoringu — používejte popisná jména jako valid_customer_age), constraint (booleovský SQL výraz nad záznamem) a akci při porušení. Ve výrazu smíte použít literály, identifikátory sloupců a deterministické vestavěné SQL funkce a operátory; zakázané jsou subquery, agregační funkce, analytické a ranking window funkce a table-valued generator funkce.

from pyspark import pipelines as dp

# dekoratory lze vrstvit — kazde pravidlo ma vlastni akci
@dp.table
@dp.expect("has_phone", "phone_number IS NOT NULL")
@dp.expect_or_drop("valid_price", "price >= 0 AND price <= 10000")
@dp.expect_or_fail("required_customer_id", "customer_id IS NOT NULL")
def validated_orders():
    return spark.readStream.table("raw.orders")

CREATE OR REFRESH STREAMING TABLE validated_orders(
    CONSTRAINT has_phone EXPECT (phone_number IS NOT NULL),
    CONSTRAINT valid_price EXPECT (price BETWEEN 0 AND 10000) ON VIOLATION DROP ROW,
    CONSTRAINT required_customer_id EXPECT (customer_id IS NOT NULL) ON VIOLATION FAIL UPDATE
) AS SELECT * FROM STREAM(raw.orders);

Pravidla nemusíte psát ke každé tabulce zvlášť. Python API umí přijmout celý slovník pravidel (@dp.expect_all_or_drop(rules)) — a ten slovník může pocházet z centrální governance tabulky v Unity Catalogu. Jedna sada pravidel označkovaná tagy, sdílená napříč pipelines:

from pyspark import pipelines as dp
from pyspark.sql.functions import col

def get_rules(tag):
    # nacte pravidla z governance tabulky podle tagu
    df = spark.read.table("governance.data_quality.rules").filter(col("tag") == tag).collect()
    return {row["name"]: row["constraint"] for row in df}

@dp.table
@dp.expect_all_or_drop(get_rules("financial"))
def validated_payments():
    return spark.readStream.table("raw.payments")

Výsledky sledujete v UI: Jobs & Pipelines → vybrat pipeline → dataset s expectations → záložka Data quality. Metriky (počet prošlých a poručených záznamů na pravidlo) jdou také dotazovat programově z event logu — základ pro vlastní dashboardy a alerty.

V roli data engineera pojišťovny ClearCover postavíte deklarativní pipeline nad denními dávkami pojistných událostí plnými chyb: NULL identifikátory, neparsovatelná data a částky, záporné hodnoty i nečekané sloupce. Přidáte expectations na nullability a rozsahy, přetypování přes col().cast() a Auto Loader s rescued data column — aby špatný záznam nikdy nedotekl k aktuárským modelům.

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