Pipelines met Lakeflow Spark Declarative Pipelines

Zeg wat je wilt, laat het platform het draaien

In les 1 en 2 bouwden we lagen met handgeschreven statements. Dat werkt voor een demo en faalt in productie: je moet zelf stappen ordenen, incrementele reads regelen, storingen opnieuw proberen en datakwaliteit vastleggen. Lakeflow Spark Declarative Pipelines leggen dat werk bij het platform. Jij declareert de datasets en de queries die ze definiëren en de pipeline bepaalt de volgorde, de incrementele verwerking en het herstel.

In deze les behandelen we de drie kernobjecten (flows, streaming tables, materialized views), expectations voor datakwaliteit en de opbouw van een medallion-pipeline. Het orkestreren van de pipeline binnen een grotere workflow is les 4.

De concepten

De Databricks-documentatie beschrijft Lakeflow Spark Declarative Pipelines als gebouwd op Apache Spark Declarative Pipelines, een declaratief framework voor batch- en streamingpipelines, met extra functionaliteit op Databricks Runtime. Je schrijft ze in SQL of Python. Drie concepten dragen het meeste gewicht.

Je kunt ook losse streaming tables en materialized views maken in Databricks SQL of Python, buiten een pipeline.

Een opmerking over namen. Dit product heette Delta Live Tables. Oudere code gebruikt import dlt en @dlt.table. De huidige Python-interface importeert de module als dp en de documentatie koppelt een streaming @dlt.table aan @dp.table en een batch-@dlt.table aan @dp.materialized_view. Als je oudere artikelen en voorbeelden leest, vertaal ze dan.

Een medallion-pipeline

De twee listings hieronder zijn schematisch. Ze tonen de structuur van een bronze-, silver- en gold-pipeline. Controleer de exacte syntaxis in de actuele referentie voor jouw runtime voor je ze overneemt.

# Schematische Python-pipeline. De importregel is de gedocumenteerde;
# controleer de referentie voor de exacte decoratoropties in jouw versie.
from pyspark import pipelines as dp
from pyspark.sql import functions as F
 
@dp.table                      # streaming table: leest nieuwe bestanden incrementeel
def orders_bronze():
    return (
        spark.readStream.format("cloudFiles")
             .option("cloudFiles.format", "json")
             .load("/Volumes/dev_sales/landing/orders/")
    )
 
@dp.table                      # streaming table: silver gebouwd uit bronze, niet uit de bron
def orders_silver():
    return (
        spark.readStream.table("orders_bronze")
             .select(F.col("order_id").cast("bigint"),
                     F.col("amount").cast("decimal(12,2)"),
                     F.col("order_ts").cast("timestamp"))
    )
 
@dp.materialized_view          # materialized view: aggregatie voor één zakelijke vraag
def orders_daily_gold():
    return (
        spark.read.table("orders_silver")
             .groupBy(F.to_date("order_ts").alias("order_date"))
             .agg(F.sum("amount").alias("revenue"))
    )

De documentatie noemt twee regels voor deze functies. Een functie die een dataset definieert, moet een Spark DataFrame teruggeven en mag binnen de definitie geen acties aanroepen zoals collect(), count() of save().

In SQL leest dezelfde pipeline als drie statements met CREATE OR REFRESH STREAMING TABLE en CREATE OR REFRESH MATERIALIZED VIEW. We houden de Python-listing als enig voorbeeld, omdat de exacte SQL-clausules voor ingestie per bron en versie verschillen.

In een pipeline heeft Auto Loader geen schemalocatie en geen checkpointlocatie nodig. De pipeline beheert beide.

Expectations: datakwaliteit als onderdeel van de definitie

Een expectation is een optionele clausule in een datasetdefinitie die elk record toetst aan een SQL-booleaanse voorwaarde. Elke expectation heeft een naam die uniek is binnen de dataset. Jij kiest wat er gebeurt als een record faalt.

Dit zijn de voorbeelden uit de documentatie:

CONSTRAINT valid_customer_age EXPECT (age BETWEEN 0 AND 120)
CONSTRAINT valid_current_page EXPECT (current_page_id IS NOT NULL AND current_page_title IS NOT NULL) ON VIOLATION DROP ROW
CONSTRAINT valid_count EXPECT (count > 0) ON VIOLATION FAIL UPDATE

@dp.expect("valid_customer_age", "age BETWEEN 0 AND 120")
@dp.expect_or_drop("valid_current_page", "current_page_id IS NOT NULL AND current_page_title IS NOT NULL")
@dp.expect_or_fail("valid_count", "count > 0")

Python biedt ook expect_all, expect_all_or_drop en expect_all_or_fail, die een dictionary met namen en voorwaarden nemen. Gegroepeerde acties zijn er niet in SQL.

Beperkingen om te kennen: metrics worden vastgelegd voor waarschuwen en weggooien, niet voor laten falen. Een voorwaarde mag geen eigen Python-functies, aanroepen naar externe services of subqueries bevatten. Voorwaarden zijn controles op één record, dus een regel die een opzoeking in een andere tabel nodig heeft, hoort in de transformatie.

De actie kiezen

Beslis per regel, samen met de dataeigenaar, en leg de keuze vast.

• Laten falen bij regels waarbij slechte data nooit bij een afnemer mag komen: een ontbrekende primaire sleutel, een negatieve hoeveelheid in een grootboek.
• Weggooien bij rijen die onbruikbaar zijn maar de batch niet fout maken, en alleen als het aantal weggegooide rijen wordt bewaakt. Stil weggooien is dataverlies.
• Waarschuwen bij regels die je nog aan het leren bent. Volg de overtredingstellingen een paar weken en bevorder de regel dan naar weggooien of laten falen.

Dit sluit de lus van les 1: de silverquery die stilletjes onbruikbare rijen verwijderde, wordt een expectation waarvan je het aantal ziet.

Opbouw en testen

• Eén pipeline per domein en omgeving. Pipelines voor dev, test, acceptatie en productie worden door dezelfde bundle uitgerold (les 5) met verschillende targets.
• Houd transformaties testbaar. Zet bedrijfslogica in gewone functies die DataFrames nemen en teruggeven en test ze zonder een pipeline te draaien.
• Behandel expectations als tests op echte data. Review ze in pull requests zoals elke andere code.
• Itereer klein. Itereer op een kleine steekproef in dev en promoveer door de omgevingen.

Samenvatting

Declareer datasets, geen stappen. Gebruik streaming tables voor incrementele lagen en materialized views voor aggregaties. Zet datakwaliteit in de definitie met expectations en beslis per regel of je waarschuwt, weggooit of laat falen.

Volgende les: jobs en orkestratie rond je pipelines.