Qualité des données dans les vues matérialisées du lac

Dans les architectures de médaillon, vous devez appliquer la qualité des données à chaque étape. Une mauvaise qualité des données peut entraîner des insights incorrects et des inefficacités opérationnelles.

Cet article explique comment implémenter des contrôles de qualité des données dans des vues de lac matérialisées (MLV) dans Microsoft Fabric.

Implémenter la qualité des données

Les contraintes au niveau des lignes valident chaque ligne isolément (par exemple, «customerName n’est pas nul sur cette ligne »). Cependant, de nombreux vrais problèmes de qualité concernent les propriétés d’une colonne dans son ensemble :

• Unicité — une clé principale ou une clé métier telle que order_id ne possède aucun doublon.
• Complétude — une colonne critique comme customer_id ne contient pas de nulls.
• Forme statistique — la moyenne, la taille ou la distribution d’une colonne reste dans une plage attendue.

Détecter ces problèmes lors de l’actualisation empêche les données erronées de se propager en aval dans votre chaîne bronze→silver→gold, et vous donne un signal de qualité auditable à chaque exécution.

• Contraintes de qualité des données au niveau des lignes
• Contrôles de qualité des données au niveau des colonnes

Contraintes de qualité des données au niveau des lignes

Dans les vues de lac matérialisées (MLV) dans Fabric, maintenez la qualité des données en définissant des contraintes sur vos vues. Sans vérifications explicites, les problèmes de données mineurs peuvent augmenter le temps de traitement ou échouer le pipeline.

Lorsqu’une ligne enfreint une contrainte, vous pouvez utiliser l’une des actions suivantes :

• FAIL : arrête l’actualisation MLV lors de la première violation de contrainte. Il s’agit du comportement par défaut, même lorsque vous ne spécifiez FAILpas .

• DROP : poursuit le traitement et supprime les enregistrements qui violent la contrainte. La vue traçabilité affiche le nombre d’enregistrements supprimés.

Définir des vérifications de qualité des données dans une vue matérialisée du lac

Lorsque vous créez une vue matérialisée du lac, vous pouvez définir des contraintes ( règles de qualité des données qui valident chaque ligne lors d’une actualisation). Une contrainte est une expression booléenne que chaque ligne doit satisfaire. Les lignes qui passent sont écrites dans la table de sortie. Les lignes qui échouent sont gérées en fonction du paramètre de violation : elles sont supprimées en mode silencieux ou provoquent l’échec de l’actualisation entière.

L’exemple suivant définit la contrainte cust_blank, qui vérifie si le customerName champ n’est pas null. La contrainte exclut les lignes avec une valeur Null customerName du traitement.

CREATE OR REPLACE MATERIALIZED LAKE VIEW IF NOT EXISTS silver.customers_enriched  
(CONSTRAINT cust_blank CHECK (customerName is not null) on MISMATCH DROP)
AS
SELECT
    c.customerID,
    c.customerName,
    c.contact, 
    CASE  
       WHEN COUNT(o.orderID) OVER (PARTITION BY c.customerID) > 0 THEN TRUE  
       ELSE FALSE  
    END AS has_orders 
FROM bronze.customers c LEFT JOIN bronze.orders o 
ON c.customerID = o.customerID;

Fonctions intégrées au système

Les fonctions Spark/SQL intégrées telles que UPPER(), LOWER(), TRIM(), COALESCE(), INITCAP() et DATE_FORMAT() sont entièrement prises en charge dans tous les contextes MLV pour l’actualisation CREATE et LINEAGE.

CREATE MATERIALIZED LAKE VIEW sample_lakehouse.silver.names (
CONSTRAINT substring_check
CHECK (SUBSTRING(name, 1, 2) = 'Al') ON MISMATCH drop
) AS
SELECT id, name
FROM (VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Ann')) AS t(id, name)

UDFs – Définis et enregistrés dans le même notebook

Les UDFs enregistrées avec spark.udf.register() dans le même notebook sont prises en charge pour l'instruction CREATE dans tous les contextes. Pour l’actualisation LINEAGE, seuls les contextes PySpark sont pris en charge, car la définition UDF s’exécute dans le cadre de l’exécution planifiée du notebook.

from pyspark.sql import SparkSession
from pyspark.sql.types import StringType
from pyspark.sql.functions import expr

spark = SparkSession.builder.getOrCreate()

# UDF: Extract domain from email
def extract_email_domain(email):
    if email is None or '@' not in email:
        return None
    return email.split('@')[1]

# Registration
spark.udf.register(
    "udf_email_domain",
    extract_email_domain,
    StringType()
)

@fmlv.materialized_lake_view(
    name="udf_testing_silver.mlv_high_value_customers",
    comment="High-value customers identified by UDF criteria",
    table_properties={"delta.enableChangeDataFeed": "true"}
)
def mlv_high_value_customers():
    return spark.sql("""
        SELECT 
            c.customer_id,
            c.name,
            c.email,
            udf_email_domain(c.email) as email_domain,
            c.segment,
            c.lifetime_value,
            total_transactions.total_amount,
            total_transactions.txn_count
        FROM udf_testing_bronze.customers c
        INNER JOIN (
            SELECT 
                customer_id,
                SUM(amount) as total_amount,
                COUNT(*) as txn_count
            FROM udf_testing_bronze.transactions
            WHERE udf_is_positive(amount)
            GROUP BY customer_id
            HAVING SUM(amount) > 1000
        ) total_transactions ON c.customer_id = total_transactions.customer_id
        WHERE udf_validate_customer(c.email, c.age)
            AND c.segment IN ('premium', 'vip')
    """)

print("✓ Created mlv_high_value_customers")

Bibliothèques tierces - Pandas UDFs

Les bibliothèques tierces telles que les UDF Pandas permettent d’implémenter des règles de qualité des données avec un traitement vectorisé. Ils permettent des validations avancées telles que la logique métier personnalisée, les vérifications statistiques ou la détection de modèles qui ne sont pas possibles avec les fonctions intégrées. Cela permet de créer des contraintes de qualité des données évolutives et réutilisables lors de la création et de l’actualisation de MLV.

  import fmlv
  from pyspark.sql.types import StructType, StructField, IntegerType, TimestampType, StringType, DoubleType, BooleanType
  from datetime import datetime

  import pandas as pd
  from pyspark.sql.types import BooleanType
  def pandas_check_impl(val):
    # Reject if value < median of [100, 200, 300]
    return val >= pd.Series([100, 200, 300]).median()
  spark.udf.register("pandas_check", pandas_check_impl, BooleanType())

  @fmlv.materialized_lake_view(
    name="silver.pyspark_from_two_sqlmlv_inner_pandas",
    comment="PySpark MLV INNER JOIN using pandas-based constraint and DROP violations"
  )
  @fmlv.check(
    name="dq_pandas_check",
    condition="pandas_check(l3)",
    action="DROP"
  )
  def pyspark_from_two_sqlmlv_inner_pandas():
    # Define the function

    # Register the function as a Spark UDF

    # Read source tables
    df1 = spark.table("silver.base_sqlmlv")
    df2 = spark.table("silver.base_sqlmlv")

    # Rename columns for unique join
    df_left = df1.select([df1[col].alias(f"l{i+1}") for i, col in enumerate(df1.columns)])
    df_right = df2.select([df2[col].alias(f"r{i+1}") for i, col in enumerate(df2.columns)])

    # Perform INNER JOIN
    df = df_left.join(df_right, df_left.l1 == df_right.r1, "inner")
    return df


  df = spark.table("silver.pyspark_from_two_sqlmlv_inner_pandas")
  # All amounts should be >= median (200)
  assert all(df.select("l3").rdd.map(lambda r: r[0] >= 200).collect()), "Unexpected low-value rows found"
  print("PySpark MLV INNER JOIN with pandas-based DQ DROP passed")

Bibliothèques personnalisées – Python wheel (.whl)

Les fonctions empaquetées en tant que fichiers jar ou fichiers de roue peuvent être installées sur le cluster Fabric (via les paramètres d’environnement) et utilisées dans les définitions MLV. CREATE et LINEAGE REFRESH sont pris en charge pour les contextes PySpark. Pour plus d’informations, consultez Gérer les bibliothèques personnalisées dans les environnements Fabric .

  %%pyspark
  import fmlv
  from pyspark.sql.types import StructType, StructField, IntegerType, TimestampType, StringType, DoubleType, BooleanType
  from datetime import datetime

  from pyspark.sql.types import BooleanType
  from custom_dq_lib import threshold_check
  def custom_check_impl(val):
    return threshold_check(val, threshold=200)
  spark.udf.register("custom_check", custom_check_impl, BooleanType())

  @fmlv.materialized_lake_view(
    name="silver.pyspark_from_two_sqlmlv_inner_custom_whl",
    comment="PySpark MLV INNER JOIN using custom DQ library and DROP violations",
    replace=True
  )
  @fmlv.check(
    name="dq_custom_check",
    condition="custom_check(l3)",
    action="DROP"
  )
  def pyspark_from_two_sqlmlv_inner_custom():
    # Wrap the custom function as Spark UDF


    # Read source tables
    df1 = spark.table("silver.base_sqlmlv")
    df2 = spark.table("silver.base_sqlmlv")

    # Rename columns for unique join
    df_left = df1.select([df1[col].alias(f"l{i+1}") for i, col in enumerate(df1.columns)])
    df_right = df2.select([df2[col].alias(f"r{i+1}") for i, col in enumerate(df2.columns)])

    # Perform INNER JOIN
    df = df_left.join(df_right, df_left.l1 == df_right.r1, "inner")
    return df


    df = spark.table("silver.pyspark_from_two_sqlmlv_inner_custom_whl")
    # All amounts should be >= threshold (200)
    assert all(df.select("l3").rdd.map(lambda r: r[0] >= 200).collect()), "Unexpected low-value rows found"
    print("PySpark MLV INNER JOIN with custom DQ library passed")

Fonctions de données utilisateur Fabric

Les fonctions de données utilisateur fabric (UDF) sont définies et gérées de manière centralisée dans l’espace de travail Fabric. Ils sont disponibles pour tout notebook ou pipeline sans nécessiter de réenregistrement à chaque session, ce qui les rend idéaux pour les pipelines MLV de production. Cette fonctionnalité est prise en charge uniquement dans les contextes PySpark pour la mise à jour des types CREATE et LINEAGE. En savoir plus sur les fonctions de données utilisateur ici. Pour plus d’informations, consultez vue d’ensemble des fonctions de données utilisateur Fabric.

  %%pyspark
  import fmlv
  from pyspark.sql import functions as F
  from notebookutils import udf

  myFuncs = udf.getFunctions("UserDataFunction_1")

  def add_greeting_column(df):
    pdf = df.toPandas()
    pdf["greeting"] = pdf["name"].apply(lambda n: myFuncs.hello_fabric(n))
    return spark.createDataFrame(pdf)

  import fmlv
  from pyspark.sql.types import StructType, StructField, IntegerType, StringType, DoubleType

  # -------------------------------------------------------------
  # Define base PySpark MLV (no SQL)
  # -------------------------------------------------------------
  @fmlv.materialized_lake_view(
    name="silver.base_pysparkmlv",
    comment="Base MLV created using PySpark",
    replace=True
  )
  def base_pysparkmlv():

    schema = StructType([
        StructField("id", IntegerType(), True),
        StructField("name", StringType(), True),
        StructField("amount", DoubleType(), True),
        StructField("country", StringType(), True)
    ])

    data = [
        (1, "Alice", 100.0, "US"),
        (2, "Bob", 200.0, "UK"),
        (3, "Charlie", 300.0, "UK")
    ]

    return spark.createDataFrame(data, schema)

  @fmlv.materialized_lake_view(
    name="silver.mlv_udfn_null_test",
    replace=False
  )
  @fmlv.check(
    name="null_check",
    condition="greeting IS NOT NULL",
    action="FAIL"
  )
  def mlv_udfn_null_test():
    df = spark.table("silver.base_pysparkmlv")
    return add_greeting_column(df)

Contrôles de qualité des données au niveau des colonnes

Fonctionnement

Définissez des vérifications de qualité des données au niveau des colonnes (DQ) avec trois décorateurs Python que vous placez dans une cellule de carnet, séparément de la définition MLV :

Parce que la suite cible un MLV par nom, la responsabilité de la qualité des données peut être attribuée, exécutée et planifiée indépendamment de la définition MLV — et vous pouvez effectuer des vérifications sans déclencher un rafraîchissement.

Étape 1 : Configurez l’environnement d’exécution

Configurez les dépendances pour votre version d’exécution Fabric avant de définir des contrôles de qualité des données au niveau des colonnes.

Fabric runtimes antérieurs à la version 2.0

Téléchargez la PyDeequ 1.7.0 wheel et chargez-la sous forme de bibliothèque personnalisée dans votre environnement Fabric. Pour les instructions, voir Importer des bibliothèques personnalisées.

Fabric runtime 2.0 ou version ultérieure

Ajoutez les propriétés suivantes de Spark :

Vous pouvez ajouter ces propriétés soit dans un environnement, soit dans un carnet :

• Environnement : Ajoutez les propriétés sous Spark Compute>Propriétés Spark, puis enregistrez et publiez l’environnement. Pour les instructions, voir Ajouter des configurations Spark à un environnement personnalisé.
• Notebook : Ajoutez les propriétés dans l’objet conf d’une commande %%configure dans la première cellule de code, avant de commencer la session Spark. Pour les instructions, voir la commande magique de configuration de la session Spark.

Étape 2 : Définir les contrôles au niveau de la colonne

Dans cet exemple, définissez des contrôles de qualité des données au niveau de la colonne pour la sales_db.daily_orders vue Lake matérialisée, spécifiée par mlv_name dans @fmlv.dq_suite.

Placez l’enregistrement de la suite et ses fonctions d’entrée dans une seule cellule de données de qualité. L’exemple suivant effectue deux vérifications en un seul passage de validation — customer_id doit être complet (sans nulls) et order_id doit être unique :

import os
os.environ["SPARK_VERSION"] = "3.5"   # works on both 3.5 and 4.1
import fmlv

# Entry function — returns a PyDeequ VerificationResult DataFrame
def completeness_and_uniqueness(df):
    from pydeequ.checks import Check, CheckLevel
    from pydeequ.verification import VerificationSuite, VerificationResult

    check = Check(spark, CheckLevel.Error, "Completeness & Uniqueness")
    result = (
        VerificationSuite(spark)
        .onData(df)
        .addCheck(
            check.isComplete("customer_id")   # check 1: no nulls
                 .isUnique("order_id")        # check 2: no duplicates
        )
        .run()
    )
    return VerificationResult.checkResultsAsDataFrame(spark, result)
# Data quality cell — separate from the MLV definition
@fmlv.dq_suite(mlv_name="sales_db.daily_orders", replace=True)
@fmlv.validate("completeness_and_uniqueness", "FAIL")
def daily_orders_dq():
    pass

Ajoutez des métriques sans faire échouer l’exécution. Empilez un @fmlv.analyze pass pour calculer les métriques de colonnes (taille, complétude, moyenne, ...) pour l’observabilité. Les résultats d’analyse sont toujours enregistrés et ne bloquent jamais une exécution — utile tant que les règles de validation évoluent encore.

import os
os.environ["SPARK_VERSION"] = "3.5"   # works on both 3.5 and 4.1
import fmlv

# Entry function — returns a PyDeequ AnalyzerContext metrics DataFrame
def column_profiling(df):
    from pydeequ.analyzers import AnalysisRunner, AnalyzerContext, Size, Completeness, Mean
    result = (
        AnalysisRunner(spark)
        .onData(df)
        .addAnalyzer(Size())
        .addAnalyzer(Completeness("customer_id"))
        .addAnalyzer(Mean("amount"))
        .run()
    )
    return AnalyzerContext.successMetricsAsDataFrame(spark, result)
# Data quality cell — separate from the MLV definition
@fmlv.dq_suite(mlv_name="sales_db.daily_orders", replace=True)
@fmlv.validate("completeness_and_uniqueness", "FAIL")
@fmlv.analyze("column_profiling")
def daily_orders_dq():
    pass

Maîtriser le résultat: ÉCHEC vs. AVERTISSEMENT

L’action sur @fmlv.validate décide ce qui se passe lorsqu’un test échoue :

• FAIL — l’échec de la vérification fait échouer l’exécution du contrôle qualité des données et bloque les MLV en aval qui en dépendent. Utilisez-le pour des contraintes strictes comme les clés primaires et les champs obligatoires.
• WARN — la défaillance est enregistrée mais l’exécution réussit tout de même et les MLV en aval ne sont pas bloquées. Utilisez-le pour des règles souples ou indicatives.

Étape 3 : Activez les contrôles de qualité des données dans une planification

Choisissez si les vérifications au niveau de la colonne s’exécutent dans le cadre d’un calendrier de rafraîchissement. Dans les paramètres avancés de l’expérience Créer un nouvel horaire , le menu déroulant des vérifications de qualité des données propose deux options :

• Vérifiez après chaque exécution — la suite qualité des données s’exécute automatiquement après chaque rafraîchissement MLV de cette planification terminé.
• Pas de vérifications — le contrôle de la qualité des données n’est pas effectué pour ce planning ; le MLV se rafraîchit toujours sur son rythme.

Le paramètre des vérifications de qualité des données dans les paramètres avancés d’un planning — choisissez Vérifier après chaque exécution ou Aucun contrôle.

Comment une vérification ayant échoué affecte le pipeline

Dans la vue de la lignée, la vérification de la qualité des données sur gold.daily_dq_reportusage était définie avec FAIL et a échoué. En conséquence, son élément dépendant en aval, gold.monthly_dq_report…, est automatiquement sauté, tandis que les branches indépendantes (gold.expected_regions, gold.region_mlv_dq_rep…) se complètent normalement.

Une FAIL vérification de la qualité des données arrête la chaîne — le dépendant en aval du nœud défaillant est sauté, mais les branches indépendantes continuent.

Ordre d’exécution. Lorsqu’une suite possède les deux types, toutes @fmlv.validate les fonctions s’exécutent d’abord (dans l’ordre des définitions), puis toutes les @fmlv.analyze fonctions. Un échec est enregistré et l’exécution continue ; le résultat global est décidé après la fin de tous les tests — l’exécution n’échoue que si une vérification d’action FAIL a échoué.

Étape 4 : Consultez et analysez le rapport sur la qualité des données

L’historique des exécutions reflète si la qualité des données était activée ou non à chaque exécution. Vous pouvez aussi exécuter une suite à la demande depuis son carnet à tout moment, indépendamment de tout rafraîchissement.

Sélectionnez n’importe quel nœud MLV dans la vue de lignage et ouvrez l’onglet Qualité des données du panneau de détails pour voir les résultats de cette exécution :

• Métriques calculées — les métriques agrégées au niveau des colonnes calculées par les analyseurs Deequ (par exemple, Completeness de region).
• Résultats de vérification — un résumé du total des contrôles, du total des contrôles réussis et du total des contrôles échoués, suivi de chaque contrainte et de son statut de réussite ou d’échec, afin que vous puissiez voir exactement quelle règle a réussi ou a échoué.

L’onglet Qualité des données affiche les métriques calculées et les résultats de vérification par contrainte pour l’exécution MLV sélectionnée.

L’historique des exécutions MLV affiche également un état de qualité des données pour chaque exécution — Succès, Échec, ou aucun indicateur lorsqu’aucune suite n’a tourné — vous pouvez donc repérer les exécutions posant problème en un coup d’œil sans ouvrir chaque nœud.

Vérifications et analyseurs pris en charge

Parce que DQ au niveau de la colonne est alimenté par la bibliothèque open source Deequ, vous pouvez utiliser tous les contrôles et analyseurs de Deequ dans vos fonctions d’entrée. Les vérifications de validation (utilisées avec @fmlv.validate) sont Check des contraintes qui produisent un résultat de réussite/échec ; les analyseurs (utilisés avec @fmlv.analyze) calculent une métrique sans résultat de réussite/échec. Les plus courants sont listés dans les tableaux suivants.

Vérifications de validation — @fmlv.validate

Analyseurs / métriques — @fmlv.analyze

La bibliothèque open source Deequ fournit ces vérifications et analyseurs. Pour la liste complète et faisant autorité, voir le projet Deequ à github.com/awslabs/deequ.

Summary

1. Créez votre MLV et notez son nom complet qualifié.
2. Dans une cellule de qualité des données distincte, enregistrez une suite avec @fmlv.dq_suite(mlv_name=…).
3. Ajouter des vérifications — empiler @fmlv.validate("<fn>", "FAIL"|"WARN") et/ou @fmlv.analyze("<fn>") et écrire les fonctions d’entrée correspondantes retournant les DataFrames PyDeequ.
4. Lancez-le — à la demande depuis le notebook, ou activez l’interrupteur de qualité des données lors de la création d’une planification.
5. Consultez les résultats dans l’onglet qualité des données de la lignée et l’historique des exécutions.

Contenu connexe

• Qualité des données dans Materialized Lake Views — Microsoft Learn
• Aperçu de Materialized Lake Views — Microsoft Learn
• Documentation DeeQu

Contenu connexe

• Actualiser les vues matérialisées du lac
• Rapports Power BI pour la qualité des données