Návrh a implementace datových pipelines

Datová pipeline není jeden skript, ale sekvence fází, kde každá staví na validovaném výstupu předchozí. Learn modul rozlišuje šest fází: ingest → clean & validate → transform → load → serve → monitor. První čtyři se přirozeně mapují na vrstvy medallion architektury, kterou znáte z předchozích kapitol — bronze uchovává surová data pro případné přepracování, silver čistí a validuje, gold agreguje pro spotřebu.

Z návrhu pořadí operací plynou tři praktická pravidla. Za prvé: ingest zachovává surovou podobu — transformace při ingestu vám vezme možnost přehrát zpracování jinak. Za druhé: čištění patří před transformace — joiny a agregace nad nevalidními daty produkují nevalidní výsledky s mnohem horší dohledatelností. Za třetí: inkrementální zpracování od začátku — full reload je pohodlný na vývoj, ale drahý v provozu.

Fáze load a serve řeší, jak data dopravit ke konzumentům: volba mezi full refresh, inkrementálním appendem a merge patternem, partitioning cílových tabulek, optimalizované views a řízení přístupu. Poslední fáze — monitoring — není „až se to dodělá": metriky délky běhu, objemů a kvality patří do návrhu od prvního dne.

Azure Databricks nabízí dvě cesty, jak pipeline napsat. Notebooky jsou procedurální: krok za krokem říkáte, jak se má co spočítat, a máte plnou kontrolu nad každým detailem — error handlingem, cachováním, pořadím operací. Deklarativní pipelines to obracejí: popíšete co má vzniknout (tabulky a jejich definice) a framework sám odvodí závislosti, pořadí spouštění, checkpointy i retry logiku. Základní jednotkou deklarativní pipeline je flow — čte ze zdroje, transformuje a zapisuje do tabulky či sinku.

Lakeflow Job je orchestrátor: skládá tasky (notebooky, SQL, pipelines, skripty…) do DAGu — orientovaného acyklického grafu, který určuje pořadí spouštění i šíření chyb. Nezávislé tasky mohou běžet paralelně (rychlejší, ale náročnější na compute — workspace navíc limituje počet souběžných tasků), závislé sériově.

O tom, kdy se task spustí vzhledem k výsledkům předchůdců, rozhoduje Run if podmínka:

Vedle Run if podmínek existují dva speciální řídicí tasky. If/else condition task větví běh podle hodnot dat — např. výraz {{tasks.quality_check.values.invalid_count}} > 100 pošle běh do nápravné větve místo do gold agregace. For each task aplikuje stejnou logiku na každý prvek pole (seznam regionů, souborů, tabulek) bez kopírování definic tasků; souběžnost iterací je konfigurovatelná.

Parametry dělají job znovupoužitelným: v tascích na ně saháte přes {{job.parameters.<nazev>}}, k dispozici jsou i dynamické reference jako {{job.trigger.time.iso_date}} (čas plánovaného spuštění) a {{backfill.iso_datetime}} pro backfill běhy. Typické kategorie: datumové parametry (plánované běhy i backfill), prostředí (dev/staging/prod) a řízení chování (velikost dávky, formát výstupu).

Odpovědnost za chyby je rozdělená: platforma sama řeší infrastrukturní problémy (výpadky uzlů clusteru), na vás zbývají chyby datové a aplikační. Na úrovni jobu máte čtyři nástroje: retry policies (počet pokusů a interval; u continuous jobů se automaticky uplatňuje exponenciální backoff), timeouty (očekávaná a maximální doba běhu — brání nekonečně visícím taskům; u streamingu navíc thresholdy backlog metrik), notifikace (e-mail, Slack, Teams, PagerDuty, webhook) a podmíněné závislosti z předchozí sekce (větev „At least one failed" = notifikace a úklid).

V notebooku doplňujete granulární úroveň: try/except nad PySparkException umí rozlišit třídu chyby a reagovat cíleně, a vlastní retry funkce pokryje volání, na která job-level retry nedosáhne:

from pyspark.errors import PySparkException
import time

try:
    df = spark.read.table("source_data")
    df.filter("amount > 0").write.mode("append").saveAsTable("target_data")
except PySparkException as e:
    # getCondition() od PySpark 4.0; na starsich runtimech getErrorClass()
    if e.getCondition() == "TABLE_OR_VIEW_NOT_FOUND":
        print(f"Source table not found: {e.getMessageParameters()['relationName']}")
    else:
        raise

# vlastni retry s exponencialnim backoffem
def run_with_retry(operation, max_retries=3, base_delay=10):
    for attempt in range(max_retries + 1):
        try:
            return operation()
        except Exception as e:
            if attempt == max_retries:
                raise
            delay = base_delay * (2 ** attempt)
            time.sleep(delay)

Klíčový koncept, bez kterého je každý retry nebezpečný, je idempotence: opakované spuštění téhož kroku nesmí vyrobit duplicitní data. Prakticky toho dosáhnete přes MERGE místo slepého INSERT, přes streaming checkpointy (exactly-once), nebo přes „delete & insert" partiotion patternu. Pro nevalidní záznamy, které nemá smysl opakovat, slouží dead-letter / karanténní tabulka — stejný pattern, jaký jsme v kapitole 9 stavěli pro typové chyby.

Nejpřímější cesta k pipeline: každou fázi napíšete jako notebook a v Jobs & Pipelines je poskládáte do jobu. Task typu Notebook dostane cestu k notebooku, compute a parametry; závislosti nastavíte polem Depends on — a DAG se vykreslí sám. Typický výsledek je přesně trojice z labu: ingest_bronze → clean_silver → aggregate_gold.

Aby byly notebooky znovupoužitelné, parametrizujte je přes widgety a výsledky předávejte navazujícím taskům přes task values:

# precteni parametru predaneho z jobu
processing_date = dbutils.widgets.get("processing_date")
environment = dbutils.widgets.get("environment")

# predani hodnoty navazujicim taskum (napr. pro quality gate)
dbutils.jobs.taskValues.set(key="invalid_count", value=invalid_count)

# signalizace vysledku do jobu
try:
    records = process_data()
    dbutils.notebook.exit(f"SUCCESS: Processed {records} records")
except Exception as e:
    dbutils.notebook.exit(f"FAILED: {str(e)}")

Deklarativní přístup obrací perspektivu: místo orchestrace, checkpointů a závislostí deklarujete tabulky a framework se postará o zbytek. Tři hlavní přínosy: automatická orchestrace (systém odvodí pořadí z závislostí mezi tabulkami), inkrementální zpracování (streaming tables zpracují každý záznam právě jednou, materialized views přepočítávají jen změny) a vestavěné retry na úrovni Spark tasku, flow i pipeline.

Stavební kameny jsou dva. Streaming table je Delta tabulka pro append-only ingest — ideální pro čtení z cloud storage (přes Auto Loader) nebo message queues. Materialized view obsluhuje transformace — joiny, agregace — a zůstává synchronizovaná se změnami upstream tabulek.

-- streaming table: inkrementalni ingest z volume
CREATE OR REFRESH STREAMING TABLE customers_bronze
AS SELECT * FROM STREAM read_files(
  "/Volumes/main/raw_data/customers",
  format => "json"
);

-- materialized view: join synchronizovany s upstream zmenami
CREATE OR REFRESH MATERIALIZED VIEW customer_transactions
AS SELECT c.customer_id, c.customer_name, c.region,
          t.transaction_date, t.amount
FROM customers c
INNER JOIN transactions t ON c.customer_id = t.customer_id;

from pyspark import pipelines as dp

@dp.table
def customers_bronze():
    return (spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .load("/Volumes/main/raw_data/customers"))

@dp.materialized_view
def regional_sales():
    customers = spark.read.table("customers")
    transactions = spark.read.table("transactions")
    return (customers.join(transactions, on="customer_id", how="inner")
        .groupBy("region").agg({"amount": "sum"}))

Kvalitu dat v deklarativní pipeline hlídají expectations — kompletně rozebrané v kapitole 9 (sekce 9.5): EXPECT jen eviduje, ON VIOLATION DROP ROW filtruje, ON VIOLATION FAIL UPDATE zastaví flow. Vývoj podporuje Lakeflow Pipelines Editor s validací přes dry run, selektivním spouštěním, vizualizací závislostí a náhledem dat; rozsah funkcí editoru se vyvíjí, aktuální stav ověřte v dokumentaci.

Pro hotelovou platformu GlobStay (pět hotelů, noční dávky rezervací z property management systémů) postavíte medallion pipeline Bronze → Silver → Gold: parametrizované notebooky s deduplikací, NULL filtry a validací dat, pak orchestraci v Lakeflow Jobu — retry policies, quality gate s podmíněným větvením a notifikace při selhání.

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