Qualità dei dati nelle viste lake materializzate

Nelle architetture medallion è necessario applicare la qualità dei dati in ogni fase. Una scarsa qualità dei dati può portare a intuizioni errate e inefficienze operative.

Questo articolo illustra come implementare verifiche di qualità dei dati in viste lake materializzate (MLV) in Microsoft Fabric.

Implementare la qualità dei dati

I vincoli a livello di riga validano ogni riga in isolamento (ad esempio, "customerName non è nullo su questa riga"). Molti problemi reali di qualità, tuttavia, riguardano le proprietà di una colonna nel suo insieme:

• Unicità — una chiave primaria o business come order_id non ha duplicati.
• Completezza — una colonna critica come customer_id non contiene valori null.
• Profilo statistico — la media, la dimensione o la distribuzione di una colonna rimane all'interno di un intervallo atteso.

Individuare questi problemi al momento del refresh impedisce che dati errati si propaghino a valle attraverso la catena bronzo→argento→oro e ti dà un segnale di qualità auditabile ad ogni corsa.

• Vincoli di qualità dei dati a livello di riga
• Controlli di qualità dei dati a livello di colonna

Vincoli di qualità dei dati a livello di riga

Nelle materialized lake views (MLV) in Fabric, mantieni la qualità dei dati definendo vincoli sulle tue visualizzazioni. Senza controlli espliciti, i problemi di dati secondari possono aumentare il tempo di elaborazione o interrompere la pipeline.

Quando una riga viola un vincolo, è possibile usare una di queste azioni:

• FAIL: arresta l'aggiornamento MLV alla prima violazione del vincolo. Si tratta del comportamento predefinito, anche quando non si specifica FAIL.

• DROP: continua l'elaborazione e rimuove i record che violano il vincolo. La visualizzazione derivazione mostra il numero di record eliminati.

Definire controlli di qualità dei dati in una vista lake materializzata

Quando si crea una vista lake materializzata, è possibile definire vincoli, ovvero regole di qualità dei dati che convalidano ogni riga durante un aggiornamento. Un vincolo è un'espressione booleana che ogni riga deve soddisfare. Le righe che passano vengono scritte nella tabella dei risultati. Le righe che hanno esito negativo vengono gestite in base all'impostazione di violazione: vengono eliminate senza notifica o provocano un errore nel processo di aggiornamento completo.

Nell'esempio seguente viene definito il vincolo cust_blank, che controlla se il customerName campo non è Null. Il vincolo esclude le righe con un valore Null customerName dall'elaborazione.

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;

Funzioni integrate di sistema

Le funzioni Spark/SQL predefinite, ad esempio UPPER(), LOWER(), TRIM(), COALESCE(), INITCAP() e DATE_FORMAT() sono completamente supportate in tutti i contesti MLV per l'aggiornamento CREATE e 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)

UDF – Definiti e registrati nello stesso notebook

Le funzioni definite dall'utente registrate con spark.udf.register() nello stesso notebook sono supportate per CREATE in tutti i contesti. Per l'aggiornamento LINEAGE, sono supportati solo i contesti PySpark perché la definizione della UDF viene eseguita come parte dell'esecuzione programmata del 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")

Librerie di terze parti - Pandas UDF

Le librerie di terze parti come Pandas UDF consentono l'implementazione di regole di qualità dei dati mediante elaborazione vettorializzata. Abilitano convalide avanzate, ad esempio la logica di business personalizzata, i controlli statistici o il rilevamento dei criteri che non sono possibili con le funzioni predefinite. Ciò consente di creare vincoli di qualità dei dati scalabili e riutilizzabili durante la creazione e l'aggiornamento di 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")

Librerie personalizzate – Python wheel (.whl)

Le funzioni impacchettate come file JAR o wheel possono essere installate nel cluster Fabric (tramite le impostazioni dell'ambiente) e usate nelle definizioni MLV. CREATE e LINEAGE REFRESH sono supportati per i contesti PySpark. Per ulteriori dettagli, vedere Gestire librerie personalizzate in ambienti 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")

Funzioni dati utente di Fabric

Le User Data Functions (UDF) di Fabric sono definite e gestite centralmente nel workspace di Fabric. Sono disponibili per qualsiasi notebook o pipeline senza dover essere nuovamente registrati per ogni sessione, rendendoli ideali per le pipeline MLV di produzione. Questa funzionalità è supportata solo nei contesti PySpark per l'aggiornamento CREATE e LINEAGE. Altre informazioni sulle funzioni dati utente sono disponibili qui. Per altre informazioni, vedere Panoramica di Funzioni dati utente di 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)

Controlli di qualità dei dati a livello di colonna

Funzionamento

Definisci controlli di qualità dati a livello di colonna (DQ) con tre decoratori Python che posizioni in una cella di notebook, separati dalla definizione MLV:

Poiché la suite mira a un MLV per nome, la qualità dei dati può avere un responsabile, essere eseguita e programmata indipendentemente dalla definizione MLV — e puoi eseguire controlli senza attivare un aggiornamento.

Passo 1: Imposta il runtime

Configura le dipendenze per la versione runtime Fabric prima di definire controlli di qualità dati a livello di colonna.

Fabric runtime precedenti alla versione 2.0

Scarica il wheel PyDeequ 1.7.0 e caricalo come libreria personalizzata nel tuo ambiente Fabric. Per istruzioni, vedi Carica librerie personalizzate.

Fabric runtime 2.0 o versione successiva

Aggiungi le seguenti proprietà di Spark:

Puoi aggiungere queste proprietà sia in un ambiente che in un notebook:

• Ambiente: Aggiungi le proprietà sotto Spark Compute>Proprietà di Spark, poi salva e pubblica l'ambiente. Per le istruzioni, vedi Aggiungi configurazioni di Spark a un ambiente personalizzato.
• Notebook: Aggiungi le proprietà nell'oggetto conf di un comando %%configure nella prima cella di codice, prima di iniziare la sessione Spark. Per le istruzioni, consulta il comando magico di configurazione della sessione Spark.

Passo 2: Definire controlli a livello di colonna

In questo esempio, definiamo controlli di qualità dei dati a livello di colonna per la sales_db.daily_orders vista lake materializzata, specificati da mlv_name in @fmlv.dq_suite.

Metti la registrazione della suite e le sue funzioni di ingresso in un'unica cella Data Quality. Il seguente esempio esegue due controlli in un passaggio di validazione — customer_id deve essere completo (senza nulli) e order_id deve essere unico:

# 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

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

Aggiungi metriche senza causare il fallimento dell'esecuzione. Aggiungi un @fmlv.analyze passaggio per calcolare le metriche della colonna (dimensione, completezza, media, ...) per l'osservabilità. I risultati dell'analisi vengono sempre registrati e non bloccano mai una run — utile mentre le regole di validazione sono ancora in evoluzione.

@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

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)

Controlla l'esito: FAIL vs. WARN

L'azione su @fmlv.validate decide cosa succede quando un controllo fallisce:

• FAIL — il fallimento del controllo fallisce l'esecuzione della qualità dei dati e blocca le MLV a valle che ne dipendono. Usalo per vincoli rigidi come chiavi primarie e campi obbligatori.
• WARN — il guasto viene registrato ma la run ha comunque successo e le MLV a valle non vengono bloccate. Usalo per regole soft o di avviso.

Passo 3: Abilita i controlli di qualità dei dati in una pianificazione

Scegli se i controlli a livello di colonna vengono eseguiti come parte di un programma di aggiornamento. Nelle impostazioni avanzate di Crea nuova pianificazione, il menu a tendina dei controlli di qualità dei dati offre due opzioni:

• Controlla dopo ogni esecuzione — la suite di qualità dati si attiva automaticamente dopo che si completa ogni aggiornamento MLV in quella pianificazione.
• Nessun controllo — il controllo della qualità dei dati viene saltato per quel calendario; l'MLV si aggiorna comunque secondo la propria cadenza.

L'impostazione dei controlli di qualità dei dati nelle impostazioni avanzate di una pianificazione — scegli Verifica dopo ogni esecuzione oppure Nessun controllo.

Come un controllo fallito influisce sulla pipeline

Nella vista di lineage, il controllo qualità dei dati su gold.daily_dq_reportusage è stato definito con FAIL e non è riuscito. Di conseguenza, il suo dipendente a valle, gold.monthly_dq_report…, viene automaticamente ignorato, mentre i rami indipendenti (gold.expected_regions, gold.region_mlv_dq_rep…) si completano normalmente.

Un FAIL controllo qualità dei dati interrompe la catena — il nodo dipendente a valle del nodo non riuscito viene ignorato, ma i rami indipendenti continuano.

Ordine di esecuzione. Quando una suite ha entrambi i tipi, tutte @fmlv.validate le funzioni vengono eseguite prima (in ordine di definizione), poi tutte @fmlv.analyze le funzioni. Un controllo fallito viene registrato e l'esecuzione continua; l'esito complessivo viene deciso dopo che tutti i controlli sono completati — l'esecuzione fallisce solo se un controllo FAIL-action fallisce.

Passo 4: Visualizza e analizza il rapporto sulla qualità dei dati

La cronologia delle run riflette se la qualità dei dati era attiva o disattivata per ogni esecuzione. Puoi anche eseguire una suite on demand dal suo notebook in qualsiasi momento, indipendentemente da qualsiasi refresh.

Seleziona qualsiasi nodo MLV nella vista lineage e apri la scheda Qualità Dati nel pannello dettagli per vedere i risultati di quella run:

• Metriche calcolate — le metriche aggregate a livello di colonna calcolate dagli analizzatori Deequ (ad esempio, Completeness di region).
• Risultati di verifica — un riassunto dei controlli totali, del totale superato e del totale fallito, seguito da ogni vincolo e dal suo stato di superato/fallito, così puoi vedere esattamente quale regola è passata o fallita.

La scheda qualità dei dati mostra metriche calcolate e risultati di verifica per vincolo per la run MLV selezionata.

La cronologia delle run MLV mostra anche uno stato di qualità dei dati per ogni run — Successo, Fallito o nessun indicatore quando nessuna suite è stata eseguita — così puoi individuare le run problematiche a colpo d'occhio senza aprire ogni nodo.

Controlli e analizzatori supportati

Poiché il DQ a livello di colonna è gestito da Amazon Deequ, puoi usare l'intera libreria Deequ all'interno delle funzioni di ingresso. I controlli di validazione (usati con @fmlv.validate) sono Check vincoli che producono un esito di superamento/fallimento; gli analizzatori (usati con @fmlv.analyze) calcolano una metrica senza pass/fail. I più comuni sono elencati nelle tabelle seguenti.

Controlli di validazione — @fmlv.validate

Analizzatori / metriche — @fmlv.analyze

La libreria open source Amazon Deequ fornisce questi controlli e analizzatori. Per l'elenco completo e autorevole, vedi il progetto Deequ a github.com/awslabs/deequ.

Summary

1. Crea il tuo MLV e annota il suo nome pienamente qualificato.
2. In una cella separata per la qualità dei dati, registra una suite con @fmlv.dq_suite(mlv_name=…).
3. Aggiungi controlli — impila @fmlv.validate("<fn>", "FAIL"|"WARN") e/o @fmlv.analyze("<fn>") e scrivi le corrispondenti funzioni di ingresso che restituiscono i DataFrame di PyDeequ.
4. Eseguilo — su richiesta dal notebook, oppure attiva l'interruttore della qualità dei dati quando crei una pianificazione.
5. Rivedi i risultati nella scheda Data quality in Lineage e nella cronologia delle esecuzioni.

Contenuti correlati

• Qualità dei dati in Materialized Lake Views — Microsoft Learn
• Panoramica di Materialized Lake View — Microsoft Learn
• Documentazione DeeQu

Contenuti correlati

• Aggiornare le viste del lago materializzate
• Report di Power BI per la qualità dei dati