Datakvalitet i materialiserede lakevisninger

I medallion-arkitekturer skal man sikre datakvalitet på alle trin. Dårlig datakvalitet kan føre til forkert indsigt og driftsmæssig ineffektivitet.

Denne artikel forklarer, hvordan man implementerer datakvalitetskontroller i materialiserede lake views (MLV'er) i Microsoft Fabric.

Implementer datakvalitet

Rækkeniveau-begrænsninger validerer hver række isoleret (for eksempel er "customerName ikke null på denne række"). Mange reelle kvalitetsproblemer er dog egenskaber ved en kolonne som helhed:

• Unikhed — en primær eller forretningsnøgle som for eksempel order_id har ingen dubletter.
• Fuldstændighed — en kritisk kolonne som for customer_id eksempel indeholder ikke nuller.
• Statistisk form — gennemsnittet, størrelsen eller fordelingen af en søjle forbliver inden for et forventet interval.

At opdage disse problemer ved opdateringspunktet forhindrer dårlige data i at sprede sig nedstrøms gennem din bronze→sølv→guldkæde og giver dig et auditable kvalitetssignal ved hver gennemspilning.

• Rækkeniveau-datakvalitetsbegrænsninger
• Kolonneniveau datakvalitetskontroller

Rækkeniveau-datakvalitetsbegrænsninger

I materialiserede søudsigter (MLV'er) i Fabric skal du opretholde datakvaliteten ved at definere begrænsninger på dine visninger. Uden eksplicitte kontroller kan mindre dataproblemer øge behandlingstiden eller få pipelinen til at fejle.

Når en række overtræder en begrænsning, kan du bruge en af disse handlinger:

• FEJL: Stopper MLV-opdatering ved første overtrædelse af begrænsningen. Dette er standardadfærden, selv når du ikke specificerer FAIL.

• DROP: Fortsætter behandlingen og fjerner poster, der overtræder begrænsningen. Linjevisningen viser antallet af tabte poster.

Definér datakvalitetskontroller i et materialiseret sø-view

Når du opretter en materialiseret søvisning, kan du definere begrænsninger — datakvalitetsregler, der validerer hver række under en opdatering. En begrænsning er et boolesk udtryk, som hver række skal opfylde. Rækker, der passerer, skrives til outputtabellen. Rækker, der fejler, håndteres efter on-violation-indstillingen: de droppes enten lydløst eller får hele opdateringen til at fejle.

I følgende eksempel defineres begrænsningen cust_blank, som kontrollerer, om feltet customerName ikke er null. Begrænsningen udelukker rækker med en null customerName fra behandling.

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;

Systemindbyggede funktioner

Indbyggede Spark/SQL-funktioner som UPPER(), LOWER(), TRIM(), COALESCE(), INITCAP() og DATE_FORMAT() understøttes fuldt ud i alle MLV-sammenhænge for både CREATE og LINEAGE refresh.

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'er – Defineret og registreret i samme notebook

UDF'er registreret med spark.udf.register() i samme notebook understøttes til CREATE i alle sammenhænge. For LINEAGE-opdatering understøttes kun PySpark-kontekster, fordi UDF-definitionen kører som en del af den planlagte notebook-eksekvering.

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

Tredjepartsbiblioteker - Pandas UDF'er

Tredjepartsbiblioteker som Pandas UDF'er tillader implementering af datakvalitetsregler med vektoriseret behandling. De muliggør avancerede valideringer såsom tilpasset forretningslogik, statistiske tjek eller mønsterdetektion, som ikke er mulige med indbyggede funktioner. Dette hjælper med at opbygge skalerbare og genanvendelige datakvalitetsbegrænsninger under oprettelse og opdatering af 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")

Brugerdefinerede biblioteker – Python wheel (.whl)

Funktioner pakket som jar-filer eller hjulfiler kan installeres på Fabric-klyngen (via Miljøindstillinger) og bruges i MLV-definitioner. CREATE og LINEAGE REFRESH understøttes for PySpark-kontekster. Se Administrer brugerdefinerede biblioteker i Fabric-miljøer for flere detaljer.

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

Funktioner til brugerdata i struktur

Fabric User Data Functions (UDF'er) er centralt defineret og administreres i Fabric-arbejdsområdet. De er tilgængelige for enhver notebook eller pipeline uden at skulle genregistreres pr. session, hvilket gør dem ideelle til produktions-MLV-pipelines. Denne funktion understøttes kun i PySpark-sammenhænge for både CREATE og LINEAGE refresh. Lær mere om User Data Functions her. For mere information, se oversigt over Fabric User Data Functions.

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

Kolonneniveau datakvalitetskontroller

Sådan fungerer det

Definér kolonne-niveau datakvalitetskontrol (DQ) med tre Python-dekoratorer, som du placerer i en notesbogscelle, adskilt fra MLV-definitionen:

Fordi pakken målretter en MLV ved navn, kan datakvalitet ejes, køres og planlægges uafhængigt af MLV-definitionen — og du kan køre tjek uden at udløse en opdatering.

Trin 1: Opsæt køretiden

Konfigurer afhængighederne for din Fabric-runtime-version, før du definerer kolonneniveau-kvalitetskontroller af data.

Fabric-kørselstider tidligere end 2.0

Download PyDeequ 1.7.0-hjulet og upload det som et brugerdefineret bibliotek i dit Fabric-miljø. For instruktioner, se Upload brugerdefinerede biblioteker.

Fabric runtime 2.0 eller senere

Tilføj følgende Spark-egenskaber:

Du kan tilføje disse egenskaber enten i et miljø eller en notesbog:

• Miljø: Tilføj egenskaberne under Spark Compute>Spark-egenskaber, og gem og udgiv derefter miljøet. For instruktioner, se Tilføj Spark-konfigurationer til et brugerdefineret miljø.
• Notesbog: Tilføj egenskaberne i conf objektet for en %%configure kommando i den første kodecelle, før du starter Spark-sessionen. For instruktioner, se Spark session konfiguration magisk kommando.

Trin 2: Definér kolonneniveau-kontroller

I dette eksempel defineres kvalitetskontroller af data på kolonneniveau for den sales_db.daily_orders materialiserede søvisning, specificeret af mlv_name i @fmlv.dq_suite.

Sæt suite-registreringen og dens indtastningsfunktioner i en enkelt datakvalitetscelle. Følgende eksempel kører to kontroller i ét valideringsgennemløb — customer_id skal være komplet (ingen nulltal) og order_id skal være unik:

# 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

Tilføj målinger uden at fejle løbet. Stack et @fmlv.analyze pass for at beregne kolonnemetrikker (størrelse, fuldstændighed, gennemsnit, ...) for observabilitet. Analyser resultaterne registreres altid og blokerer aldrig et run — nyttigt, mens valideringsreglerne stadig udvikler sig.

# 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

Kontroller resultatet: FEJL vs. ADVARSEL

Handlingen på @fmlv.validate afgør, hvad der sker, når en kontrol fejler:

• FAIL — fejlen af kontrollen fejler datakvalitetskørslen og blokerer de downstream MLV'er, der er afhængige af den. Brug det til hårde kontrakter som primærnøgler og obligatoriske felter.
• WARN — fejlen logges, men kørslen lykkes stadig, og downstream MLV'er blokeres ikke . Brug det til bløde eller rådgivende regler.

Trin 3: Aktivér datakvalitetskontrol i en tidsplan

Vælg om kolonneniveau-tjek kører som en del af en opdateringsplan. I de avancerede indstillinger i Createn new schedule-oplevelsen tilbyder datakvalitetskontrolmenuen to muligheder:

• Tjek efter hver kørsel — datakvalitetssuiten kører automatisk efter hver MLV-opdatering i den tidsplan, der er fuldført.
• Ingen kontroller — datakvaliteten springes over for den tidsplan; MLV opdateres stadig på sin kadenc.

Datakvalitetskontrolindstillingen i en tidsplans avancerede indstillinger — vælg Tjek efter hver kørsel eller Nej-tjek.

Hvordan en fejlet kontrol påvirker pipelinen

I linjevisningen blev datakvalitetskontrollen defineret med gold.daily_dq_reportusageFAIL og fejlede. Som følge heraf bliver dens downstream-afhængige, gold.monthly_dq_report…, automatisk Springet over, mens uafhængige grene (gold.expected_regions, gold.region_mlv_dq_rep…) fuldføres normalt.

En FAIL datakvalitetskontrol stopper kæden — den fejlslagne nodes downstream-afhængige bliver sprunget over, men uafhængige grene fortsætter.

Henrettelsesordre. Når en suite har begge typer, kører alle @fmlv.validate funktioner først (i definitionsrækkefølge), derefter alle @fmlv.analyze funktioner. Et fejlet tjek registreres, og udførelsen fortsætter; det samlede resultat afgøres efter alle tjek er gennemført — kørslen fejler kun, hvis en FAIL-handlingskontrol fejlede.

Trin 4: Se og analyser datakvalitetsrapporten

Kørselshistorikken viser, om datakvaliteten var tændt eller ude for hver eksekvering. Du kan også køre en suite on demand fra dens notebook når som helst, uafhængigt af opdatering.

Vælg en hvilken som helst MLV-node i linjevisningen og åbn fanen Datakvalitet i detaljepanelet for at se resultaterne af den kørsel:

• Beregnede metrikker — de kolonneniveau-aggregatmetrikker beregnet af Deequ-analysatorerne (for eksempel Completeness af region).
• Verifikationsresultater — et resumé af samlede tjek, samlet antal beståede og samlede dumpede, efterfulgt af hver begrænsning og dens bestået/ikke-bestået-status, så du kan se præcis, hvilken regel der bestod eller dumpede.

Datakvalitetsfanen viser beregnede metrikker og verifikationsresultater pr. begrænsning for det valgte MLV-run.

MLV-kørselshistorikken viser også en datakvalitetsstatus for hver kørsel — Succes, Fejlet eller ingen indikator, når ingen suite kørte — så du kan spotte problemkørsler med et blik uden at åbne hver node.

Understøttede checks og analysatorer

Fordi kolonne-niveau DQ drives af det open source Deequ-bibliotek, kan du bruge alle Deequs kontroller og analysatorer i dine indtastningsfunktioner. Valideringskontroller (brugt med @fmlv.validate) er Check begrænsninger, der giver et bestået/ikke-bestået resultat; analysatorer (brugt med @fmlv.analyze) beregner en metrik uden bestået/ikke-bestået. De mest almindelige er listet i følgende tabeller.

Valideringskontroller — @fmlv.validate

Analysatorer / målinger — @fmlv.analyze

Det open source Deequ-bibliotek leverer disse kontroller og analysatorer. For den komplette og autoritative liste, se Deequ-projektet på github.com/awslabs/deequ.

Summary

1. Opret din MLV og noter dens fuldt kvalificerede navn.
2. I en separat datakvalitetscelle registreres en suite med @fmlv.dq_suite(mlv_name=…).
3. Tilføj tjek — stack @fmlv.validate("<fn>", "FAIL"|"WARN") og/eller @fmlv.analyze("<fn>") skriv de matchende indtastningsfunktioner, der returnerer PyDeequ DataFrames.
4. Kør det — på kommando fra notesbogen, eller slå datakvalitetsknappen til, når du opretter en tidsplan.
5. Gennemgå resultater i fanen for kvalitet af slægtslinjer og kørselshistorikken.

Relateret indhold

• Datakvalitet i Materialized Lake Views — Microsoft Learn
• Materialized Lake Views oversigt — Microsoft Learn
• DeeQu-dokumentation

Relateret indhold

• Opdater materialiserede søvisninger
• Power BI-rapporter for datakvalitet