Implementera verifieringskontroller

Datapipelines stöter ofta på poster med saknade värden, duplicerade identifierare eller värden som ligger utanför godkända intervall. Utan valideringskontroller sprids dessa datakvalitetsproblem nedströms, vilket orsakar felaktig analys, misslyckade rapporter och otillförlitliga affärsbeslut. Valideringskontroller fångar dessa problem vid tidpunkten för datainmatning, vilket säkerställer att endast kvalitetsdata flödar via din pipeline.

I den här lektionen får du lära dig hur du implementerar valideringskontroller för nullbarhet, datakardinalitet och intervallkontroll med förväntningar i Lakeflow-pipelines och begränsningar för Delta Lake-tabeller.

Förstå valideringsmetoder i Azure Databricks

Azure Databricks innehåller två primära mekanismer för att implementera valideringskontroller: pipelineförväntningar och tabellbegränsningar. Varje metod hanterar olika scenarier och erbjuder olika funktioner.

Förväntningar för pipelines tillämpar validering under datatransformeringar i Lakeflow-pipelines. Med förväntningar kan du varna, ta bort ogiltiga poster eller avbryta pipelinen när data bryter mot dina bearbetningsregler. Den här metoden fungerar bra för strömmande tabeller och materialiserade vyer där du behöver kvalitetskontroll i realtid.

Tabellbegränsningar tillämpar regler direkt på Delta Lake-tabeller. Begränsningar avvisar ogiltiga data vid skrivtillfället, vilket förhindrar att felaktiga poster någonsin kommer in i dina tabeller. Den här metoden passar batchbearbetning och scenarier som kräver strikta garantier för dataintegritet.

Med dessa metoder kan du verifiera tre viktiga dimensioner av datakvalitet: nullability (se till att nödvändiga värden finns), data kardinalitet (verifiera unikhet där förväntat) och intervallkontroll (bekräfta värden faller inom acceptabla gränser).

Implementera nullabilitetskontroller

Validering av nullabilitet säkerställer att obligatoriska kolumner innehåller värden. Tänk dig en kundtabell där email och customer_id alltid måste ha värden. Du kan implementera dessa kontroller med hjälp av förväntningar i din pipeline.

Använd Python och lägg till förväntningsdekorationer i definitionen av strömningstabellen.

from pyspark import pipelines as dp

@dp.table
@dp.expect_or_drop("valid_email", "email IS NOT NULL")
@dp.expect_or_drop("valid_customer_id", "customer_id IS NOT NULL")
def customers():
    return spark.readStream.table("raw.customers")

Samma validering i SQL använder CONSTRAINT -satsen:

CREATE OR REFRESH STREAMING TABLE customers(
    CONSTRAINT valid_email EXPECT (email IS NOT NULL) ON VIOLATION DROP ROW,
    CONSTRAINT valid_customer_id EXPECT (customer_id IS NOT NULL) ON VIOLATION DROP ROW
) AS SELECT * FROM STREAM(raw.customers);

För Delta Lake-tabeller utanför pipelines använder du NOT NULL begränsningar i tabelldefinitionen:

CREATE TABLE customers (
    customer_id INT NOT NULL,
    email STRING NOT NULL,
    first_name STRING,
    last_name STRING
);

Du kan också lägga till nullabilitetsbegränsningar i befintliga tabeller:

ALTER TABLE customers ALTER COLUMN email SET NOT NULL;

Verifiera datakardinalitet

Kardinalitetsverifiering säkerställer att kolumner som förväntas innehålla unika värden faktiskt gör det. Den här kontrollen är viktig för primära nycklar, transaktionsidentifierare och andra fält där dubbletter indikerar datakvalitetsproblem.

Pipelineförväntningar kan verifiera kardinalitet genom att söka efter villkor som indikerar unika problem. Du kan till exempel kontrollera att ett personnummer bara visas en gång per person:

from pyspark.sql.window import Window
from pyspark.sql.functions import count

@dp.table
@dp.expect("unique_ssn_per_person", "ssn_count = 1")
def employees():
    df = spark.table("raw.employees")
    w = Window.partitionBy("ssn")
    return df.withColumn("ssn_count", count("*").over(w))

Om du vill ha mer omfattande kardinalitetskontroller kombinerar du förväntningar med aggregeringslogik i omvandlingen:

CREATE OR REFRESH MATERIALIZED VIEW order_validation AS
SELECT 
    order_id,
    COUNT(*) as occurrence_count
FROM orders
GROUP BY order_id
HAVING COUNT(*) > 1;

Den här materialiserade vyn identifierar eventuella duplicerade order-ID:er så att du kan undersöka och lösa kardinalitetsproblem.

Delta Lake stöder primära nyckelbegränsningar som dokumenterar förväntad unikhet, även om dessa begränsningar är informationsmässiga och inte framtvingas:

CREATE TABLE orders (
    order_id INT NOT NULL,
    customer_id INT,
    order_date DATE,
    CONSTRAINT orders_pk PRIMARY KEY (order_id)
);

Tillämpa intervallkontroll

Intervallvalidering bekräftar att numeriska värden, datum och andra värden ligger inom godkända gränser. Den här kontrollen fångar upp datainmatningsfel, systemfel och integreringsproblem som genererar värden som inte ligger inom intervallet.

Definiera intervallförväntningar med hjälp av jämförelseoperatorer eller BETWEEN -satsen:

@dp.table
@dp.expect_or_fail("valid_age", "age BETWEEN 0 AND 150")
@dp.expect_or_fail("valid_salary", "salary >= 0")
@dp.expect_or_fail("valid_hire_date", "hire_date <= current_date()")
def employees():
    return spark.readStream.table("raw.employees")

Använd samma intervallkontroller i SQL:

CREATE OR REFRESH STREAMING TABLE transactions(
    CONSTRAINT valid_amount EXPECT (amount > 0) ON VIOLATION DROP ROW,
    CONSTRAINT valid_date EXPECT (transaction_date <= current_date()) ON VIOLATION DROP ROW,
    CONSTRAINT valid_quantity EXPECT (quantity BETWEEN 1 AND 10000) ON VIOLATION DROP ROW
) AS SELECT * FROM STREAM(raw.transactions);

För Delta Lake-tabeller använder du CHECK begränsningar för att framtvinga intervall vid skrivtillfället:

ALTER TABLE employees ADD CONSTRAINT valid_age CHECK (age >= 18 AND age <= 120);
ALTER TABLE transactions ADD CONSTRAINT positive_amount CHECK (amount > 0);

Intervallkontroller kan också verifiera affärsregler som kombinerar flera villkor:

@dp.expect("valid_discount", """
    discount_percent >= 0 
    AND discount_percent <= 100
    AND (discount_percent <= 50 OR customer_tier = 'PREMIUM')
""")

Denna förväntan säkerställer att rabatterna håller sig inom gränserna och framtvingar att endast premiumkunder får rabatter över 50%.

Välj åtgärder för valideringsfel

När valideringen misslyckas väljer du hur din pipeline ska svara. Varje åtgärd passar olika affärskrav och datakritiska nivåer.

Använd varningsåtgärden när du behöver insyn i datakvalitetsproblem utan att blockera dataflödet:

@dp.expect("has_phone", "phone_number IS NOT NULL")

Använd släpp när ogiltiga poster ska filtreras bort tyst:

@dp.expect_or_drop("complete_address", "street IS NOT NULL AND city IS NOT NULL")

Användning misslyckas när dataintegriteten är kritisk och ogiltiga poster är oacceptabla:

@dp.expect_or_fail("valid_account_balance", "balance >= 0")

Du kan visa mått för förväntningar i pipelinegränssnittet genom att välja en datauppsättning med förväntningar och öppna fliken Datakvalitet . Dessa mått hjälper dig att övervaka valideringspassfrekvenser och identifiera systematiska datakvalitetsproblem.

Nu när du förstår hur du implementerar valideringskontroller kan du se till att dina datapipelines upprätthåller kvalitetsstandarder för nullabilitet, kardinalitet och värdeintervall.