Implémenter des vérifications de validation

Les pipelines de données rencontrent souvent des enregistrements avec des valeurs manquantes, des identificateurs en double ou des valeurs qui se trouvent en dehors des plages acceptables. Sans vérification de validation en place, ces problèmes de qualité des données se propagent en aval, provoquant des analyses incorrectes, des rapports ayant échoué et des décisions métier non fiables. Les vérifications de validation interceptent ces problèmes au moment de l’ingestion des données, ce qui garantit que seules les données de qualité transitent par votre pipeline.

Dans cette unité, vous allez apprendre à implémenter des vérifications de validation pour la nullabilité, la cardinalité des données et la vérification des plages à l’aide des expectations des pipelines Lakeflow et des contraintes de table Delta Lake.

Comprendre les approches de validation dans Azure Databricks

Azure Databricks fournit deux mécanismes principaux pour implémenter des vérifications de validation : les attentes de pipeline et les contraintes de table. Chaque approche sert différents scénarios et offre des fonctionnalités distinctes.

Les attentes de pipeline permettent d'appliquer la validation pendant les transformations de données dans les pipelines Lakeflow. Les attentes vous permettent d’avertir, de supprimer des enregistrements non valides ou d’échouer le pipeline lorsque les données ne respectent pas vos règles. Cette approche fonctionne bien pour les tables de diffusion en continu et les vues matérialisées où vous avez besoin d’un contrôle de qualité en temps réel.

Les contraintes de table appliquent des règles directement sur les tables Delta Lake. Les contraintes rejettent les données non valides au moment de l’écriture, ce qui empêche les enregistrements incorrects d’entrer dans vos tables. Cette approche convient au traitement par lots et aux scénarios nécessitant des garanties strictes d’intégrité des données.

Avec ces approches, vous pouvez valider trois dimensions clés de la qualité des données : nullabilité (s’assurer que les valeurs requises existent), cardinalité des données (vérification de l’unicité où attendu) et vérification de plage (confirmation des valeurs comprises dans les limites acceptables).

Implémenter des vérifications de nullabilité

La validation de la nullabilité garantit que les colonnes requises contiennent des valeurs. Considérez une table client où email et customer_id doit toujours avoir des valeurs. Vous pouvez implémenter ces vérifications à l’aide des attentes dans votre pipeline.

À l’aide de Python, ajoutez des décorateurs d'attente à votre définition de table de streaming :

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

La même validation dans SQL utilise la CONSTRAINT clause :

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

Pour les tables Delta Lake en dehors des pipelines, utilisez les contraintes de NOT NULL dans la définition de la table :

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

Vous pouvez également ajouter des contraintes de nullabilité aux tables existantes :

ALTER TABLE customers ALTER COLUMN email SET NOT NULL;

Valider la cardinalité des données

La validation de la cardinalité garantit que les colonnes attendues contiennent réellement des valeurs uniques. Cette vérification est essentielle pour les clés primaires, les identificateurs de transaction et d’autres champs où les doublons indiquent des problèmes de qualité des données.

Les règles de validation du pipeline peuvent valider la cardinalité en vérifiant des conditions indiquant des problèmes d’unicité. Par exemple, vous pouvez vérifier qu’un numéro de sécurité sociale n’apparaît qu’une seule fois par personne :

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

Pour des vérifications de cardinalité plus complètes, combinez les attentes avec la logique d’agrégation dans votre transformation :

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

Cette vue matérialisée identifie les doublons d'ID de commande, ce qui vous permet d'examiner et de résoudre les problèmes de cardinalité.

Delta Lake prend en charge les contraintes de clé primaire qui documentent l’unicité attendue, bien que ces contraintes soient informationnelles et non appliquées :

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

Appliquer le contrôle de portée

La validation de plage confirme que les valeurs numériques, de date et d’autres valeurs se trouvent dans des limites acceptables. Cette vérification intercepte les erreurs d’entrée de données, les erreurs système et les problèmes d’intégration qui produisent des valeurs hors plage.

Définissez les attentes de plage à l’aide d’opérateurs de comparaison ou de la BETWEEN clause :

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

Dans SQL, appliquez les mêmes vérifications de plage :

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

Pour les tables Delta Lake, utilisez des contraintes CHECK pour faire respecter des plages au moment de l’écriture :

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

Les contrôles de plage peuvent aussi valider des règles métier combinant plusieurs conditions :

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

Cette attente garantit que les remises restent dans les limites et applique que seuls les clients premium reçoivent des remises supérieures à 50%.

Choisir des actions pour les échecs de validation

En cas d’échec de la validation, vous choisissez la façon dont votre pipeline répond. Chaque action répond à différentes exigences métier et aux niveaux de criticité des données.

Utilisez l’action d’avertissement lorsque vous avez besoin de visibilité sur les problèmes de qualité des données sans bloquer le flux de données :

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

Utilisez drop lorsque les enregistrements non valides doivent être filtrés silencieusement :

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

L’utilisation échoue lorsque l’intégrité des données est critique et que les enregistrements non valides sont inacceptables :

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

Vous pouvez afficher les métriques attendues dans l’interface utilisateur du pipeline en sélectionnant un jeu de données avec des attentes et en ouvrant l’onglet Qualité des données . Ces métriques vous aident à surveiller les taux de réussite de validation et à identifier les problèmes de qualité des données systématiques.

Maintenant que vous comprenez comment implémenter des vérifications de validation, vous pouvez vous assurer que vos pipelines de données conservent des normes de qualité pour la nullabilité, la cardinalité et les plages de valeurs.