Validatiecontroles implementeren

Gegevenspijplijnen komen vaak records tegen met ontbrekende waarden, dubbele identificatoren of waarden die buiten acceptabele waardenbereiken vallen. Zonder validatiecontroles worden deze problemen met gegevenskwaliteit downstream doorgegeven, waardoor onjuiste analyses, mislukte rapporten en onbetrouwbare zakelijke beslissingen worden genomen. Validatiecontroles vangen deze problemen op het moment van gegevensopname, zodat alleen kwaliteitsgegevens via uw pijplijn stromen.

In deze les leert u hoe u validatiecontroles implementeert voor null-waarden, gegevenskardinaliteit en bereikcontrole met behulp van verwachtingen van Lakeflow-pijplijnen en beperkingen voor Delta Lake-tabellen.

Inzicht in validatiemethoden in Azure Databricks

Azure Databricks biedt twee primaire mechanismen voor het implementeren van validatiecontroles: verwachtingen voor pijplijnen en tabelbeperkingen. Elke benadering dient voor verschillende scenario's en biedt verschillende mogelijkheden.

Verwachtingen voor pijplijnen passen validatie toe tijdens datatransformaties in Lakeflow-pijplijnen. Met verwachtingen kunt u waarschuwen, ongeldige records negeren of de pijplijn laten mislukken wanneer gegevens uw regels overtreden. Deze benadering werkt goed voor het streamen van tabellen en gerealiseerde weergaven waar u realtime kwaliteitscontrole nodig hebt.

Tabelbeperkingen dwingen regels rechtstreeks af op Delta Lake-tabellen. Beperkingen weigeren ongeldige gegevens tijdens het schrijven, zodat ongeldige records nooit uw tabellen binnenkomen. Deze aanpak past bij batchverwerking en scenario's waarvoor strikte garanties voor gegevensintegriteit zijn vereist.

Met deze benaderingen kunt u drie belangrijke dimensies van de gegevenskwaliteit valideren: nullability (controleren dat vereiste waarden bestaan), gegevenskardinaliteit (uniekheid verifiëren waar verwacht) en bereikcontrole (controleren dat waarden binnen acceptabele grenzen vallen).

Nullcontrole implementeren

Validatie van nullability zorgt ervoor dat verplichte kolommen waarden bevatten. Houd rekening met een klanttabel waarbij email en customer_id altijd waarden moeten hebben. U kunt deze controles implementeren met behulp van verwachtingen in uw pijplijn.

Voeg met behulp van Python expectatiedecorators toe aan de definitie van uw streamingtabel.

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")

Dezelfde validatie in SQL maakt gebruik van de CONSTRAINT component:

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);

Gebruik beperkingen in de tabeldefinitie voor Delta Lake-tabellen buiten pijplijnen NOT NULL :

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

U kunt ook beperkingen voor null-functionaliteit toevoegen aan bestaande tabellen:

ALTER TABLE customers ALTER COLUMN email SET NOT NULL;

Gegevenskardinaliteit valideren

Validatie van kardinaliteit zorgt ervoor dat kolommen die naar verwachting unieke waarden bevatten, dit daadwerkelijk doen. Deze controle is essentieel voor primaire sleutels, transactie-id's en andere velden waarbij dubbele waarden duiden op problemen met de gegevenskwaliteit.

Pijplijnwachtingen kunnen kardinaliteit valideren door te controleren op voorwaarden die duiden op problemen met uniekheid. U kunt bijvoorbeeld controleren of een burgerservicenummer slechts eenmaal per persoon wordt weergegeven:

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))

Voor uitgebreidere kardinaliteitscontroles combineert u verwachtingen met aggregatielogica in uw transformatie:

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

Deze gerealiseerde weergave identificeert eventuele dubbele order-id's, zodat u kardinaliteitsproblemen kunt onderzoeken en oplossen.

Delta Lake biedt ondersteuning voor primaire sleutelbeperkingen die de verwachte uniekheid documenteren, hoewel deze beperkingen alleen informatief zijn en niet worden afgedwongen.

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

Bereikcontrole toepassen

Met bereikvalidatie wordt bevestigd dat numerieke, datum- en andere waarden binnen acceptabele grenzen vallen. Met deze controle worden fouten bij gegevensinvoer, systeemstoringen en integratieproblemen onderschept die buiten bereikwaarden produceren.

Bereikverwachtingen definiëren met behulp van vergelijkingsoperatoren of de BETWEEN component:

@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")

Pas in SQL dezelfde bereikcontroles toe:

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);

Gebruik voor Delta Lake-tabellen CHECK beperkingen om bereikwaarden af te dwingen tijdens het schrijven.

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

Bereikcontroles kunnen ook bedrijfsregels valideren die meerdere voorwaarden combineren:

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

Deze verwachting zorgt ervoor dat kortingen binnen de grenzen blijven en dwingt af dat alleen premium klanten kortingen ontvangen van meer dan 50%.

Acties kiezen voor validatiefouten

Wanneer de validatie mislukt, kiest u hoe uw pijplijn reageert. Elke actie past bij verschillende bedrijfsvereisten en gegevenskritiekniveaus.

Gebruik de waarschuwingsactie wanneer u inzicht nodig hebt in problemen met de kwaliteit van gegevens zonder dat de gegevensstroom wordt geblokkeerd:

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

Gebruik verwijderen wanneer ongeldige records op de achtergrond moeten worden gefilterd:

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

Gebruik mislukt wanneer gegevensintegriteit kritiek is en ongeldige records onaanvaardbaar zijn:

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

U kunt metrische verwachtingen weergeven in de gebruikersinterface van de pijplijn door een gegevensset met verwachtingen te selecteren en het tabblad Gegevenskwaliteit te openen. Deze metrische gegevens helpen u bij het bewaken van validatiepaspercentages en het identificeren van systematische problemen met de kwaliteit van gegevens.

Nu u begrijpt hoe u validatiecontroles implementeert, kunt u ervoor zorgen dat uw gegevenspijplijnen kwaliteitsstandaarden onderhouden voor null-waarden, kardinaliteit en waardebereiken.