Datakvalitet i materialisert utsikt over innsjøen

I medaljongarkitekturer må du håndheve datakvalitet på alle trinn. Dårlig datakvalitet kan føre til feil innsikt og operasjonell ineffektivitet.

Denne artikkelen forklarer hvordan man implementerer datakvalitetskontroller i materialiserte lake views (MLV-er) i Microsoft Fabric.

Implementere datakvalitet

Radnivåbegrensninger validerer hver rad isolert (for eksempel, "customerName er ikke null på denne raden"). Mange reelle kvalitetsproblemer er imidlertid egenskaper ved en kolonne som helhet:

• Unikhet — en primær- eller forretningsnøkkel som order_id ikke har duplikater.
• Fullstendighet — en kritisk kolonne som for eksempel customer_id inneholder ikke nulltall.
• Statistisk form — gjennomsnittet, størrelsen eller fordelingen av en kolonne holder seg innenfor et forventet område.

Å oppdage disse problemene ved oppdateringspunktet stopper dårlig data fra å spre seg nedstrøms gjennom bronse→sølv→gullkjeden, og gir deg et reviderbart kvalitetssignal på hver runde.

• Begrensninger for rad-nivå datakvalitet
• Kolonnenivå-kvalitetskontroller av data

Begrensninger for rad-nivå datakvalitet

I materialiserte lake views (MLV-er) i Fabric, oppretthold datakvaliteten ved å definere begrensninger på dine visninger. Uten eksplisitte kontroller kan mindre dataproblemer øke behandlingstiden eller få pipelinen til å feile.

Når en rad bryter en begrensning, kan du bruke en av disse handlingene:

• FAIL: Stopper MLV-oppdatering ved første brudd på begrensningen. Dette er standardoppførselen, selv når du ikke spesifiserer FAIL.

• DROP: Fortsetter behandlingen og fjerner poster som bryter begrensningen. Linjevisningen viser antall droppede poster.

Definer datakvalitetskontroller i en materialisert innsjøvisning

Når du lager en materialisert innsjøvisning, kan du definere begrensninger — datakvalitetsregler som validerer hver rad under en oppdatering. En begrensning er et boolsk uttrykk som hver rad må oppfylle. Radene som passerer skrives til utdatatabellen. Rader som feiler håndteres i henhold til on-viol-innstillingen: de droppes enten stille eller fører til at hele oppdateringen feiler.

Følgende eksempel definerer betingelsen cust_blank, som kontrollerer om customerName feltet ikke er null. Begrensningen utelukker rader med 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;

Systemets innebygde funksjoner

Innebygde Spark/SQL-funksjoner som UPPER(), LOWER(), TRIM(), COALESCE(), INITCAP() og DATE_FORMAT() støttes fullt ut i alle MLV-kontekster 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 – Definert og registrert i samme notatbok

UDF-er registrert med spark.udf.register() i samme notatbok støttes for CREATE i alle kontekster. For LINEAGE-oppdatering støttes kun PySpark-kontekster fordi UDF-definisjonen kjører som en del av den planlagte notatbokutførelsen.

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 gjør det mulig å implementere datakvalitetsregler med vektorisert behandling. De muliggjør avanserte valideringer som tilpasset forretningslogikk, statistiske kontroller eller mønstergjenkjenning som ikke er mulig med innebygde funksjoner. Dette bidrar til å bygge skalerbare og gjenbrukbare datakvalitetsbegrensninger under MLV-opprettelse og oppdatering.

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

Egendefinerte biblioteker – Python wheel (.whl)

Funksjoner pakket som jar-filer eller hjulfiler kan installeres i Fabric-klyngen (via miljøinnstillinger) og brukes i MLV-definisjoner. CREATE og LINEAGE REFRESH støttes for PySpark-kontekster. Se Administrer tilpassede biblioteker i Fabric-miljøer for mer informasjon.

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

Funksjoner for brukerdata for stoff

Fabric User Data Functions (UDF) er sentralt definert og administreres i Fabric-arbeidsområdet. De er tilgjengelige for enhver bærbar PC eller pipeline uten å måtte registreres på nytt per økt, noe som gjør dem ideelle for produksjons-MLV-pipelines. Denne funksjonen støttes kun i PySpark-kontekster for både CREATE og LINEAGE refresh. Lær mer om User Data Functions her. For mer informasjon, se oversikt 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)

Kolonnenivå-kvalitetskontroller av data

Slik fungerer det

Definer kolonnenivå-datakvalitetskontroller (DQ) med tre Python-dekoratorer som du plasserer i en notatblokkcelle, adskilt fra MLV-definisjonen:

Fordi pakken retter seg mot en MLV med navn, kan datakvalitet eies, kjøres og planlegges uavhengig av MLV-definisjonen — og du kan kjøre sjekker uten å utløse en oppdatering.

Trinn 1: Sett opp kjøretiden

Konfigurer avhengighetene for din Fabric-kjøretidsversjon før du definerer kolonnenivå-kvalitetskontroller av data.

Fabric-kjøretider tidligere enn 2.0

Last ned PyDeequ 1.7.0-hjulet og last det opp som et egendefinert bibliotek i ditt Fabric-miljø. For instruksjoner, se Last opp tilpassede biblioteker.

Fabric kjøretid 2.0 eller nyere

Legg til følgende Spark-egenskaper:

Du kan legge til disse egenskapene enten i et miljø eller en notatbok:

• Miljø: Legg til egenskapene under Spark Compute>Spark-egenskaper, lagre og publiser deretter miljøet. For instruksjoner, se Legg til Spark-konfigurasjoner i et egendefinert miljø.
• Notatbok: Legg til egenskapene i objektet conf til en %%configure kommando i den første kodecellen før du starter Spark-økten. For instruksjoner, se Spark session configuration magic command.

Trinn 2: Definer kolonnenivå-kontroller

I dette eksempelet, definer kolonnenivå-kvalitetskontroller for den sales_db.daily_orders materialiserte innsjøvisningen, spesifisert av mlv_name i @fmlv.dq_suite.

Samle suite-registreringen og dens inntastingsfunksjoner i én enkelt datakvalitetscelle. Følgende eksempel kjører to kontroller i én valideringsrunde — customer_id må være komplett (ingen nulltall) og order_id må være unikt:

# 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

Legg til målinger uten å feile løpet. Stable et @fmlv.analyze pass for å beregne kolonnemetrikker (størrelse, fullstendighet, gjennomsnitt, ...) for observabilitet. Analyse av resultater registreres alltid og blokkerer aldri en kjøring — nyttig mens valideringsreglene fortsatt er under utvikling.

# 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 utfallet: FAIL vs. WARNING

Handlingen på @fmlv.validate avgjør hva som skjer når en sjekk feiler:

• FAIL — sjekkfeilen feiler datakvalitetskjøringen og blokkerer de nedstrøms MLV-ene som er avhengige av den. Bruk det til harde kontrakter som primærnøkler og obligatoriske felt.
• WARN — feilen logges, men kjøringen lykkes fortsatt, og nedstrøms MLV-er blokkeres ikke . Bruk det til myke eller rådgivende regler.

Trinn 3: Aktiver datakvalitetskontroller i en tidsplan

Velg om kolonnenivå-sjekker skal kjøres som en del av en oppdateringsplan. I de avanserte innstillingene i Opprett ny plan-opplevelsen tilbyr nedtrekksmenyen for datakvalitetskontroller to alternativer:

• Sjekk etter hver kjøring — datakvalitetssuiten kjører automatisk etter hver MLV-oppdatering i den planen er fullført.
• Ingen sjekker — datakvaliteten hoppes over for den planen; MLV oppdaterer fortsatt sin rytme.

Datakvalitetssjekk-innstillingen i en tidsplans avanserte innstillinger — velg Sjekk etter hver kjøring eller Nei-sjekker.

Hvordan en mislykket sjekk påvirker pipelinen

I lineage-visningen ble datakvalitetskontrollen definert gold.daily_dq_reportusage med FAIL og feilet. Som et resultat blir dens nedstrøms avhengige, gold.monthly_dq_report…, automatisk hoppet over, mens uavhengige grener (gold.expected_regions, gold.region_mlv_dq_rep…) fullføres normalt.

En FAIL datakvalitetskontroll stopper kjeden — den feilede nodens nedstrøms avhengige blir hoppet over, men uavhengige grener fortsetter.

Henrettelsesordre. Når en suite har begge typer, kjøres alle @fmlv.validate funksjoner først (i definisjonsrekkefølge), deretter alle @fmlv.analyze funksjoner. En feilsjekk registreres og utførelsen fortsetter; det overordnede utfallet avgjøres etter at alle sjekker er fullført — kjøringen feiler bare hvis en FAIL-handlingssjekk feilet.

Trinn 4: Se og analyser datakvalitetsrapporten

Kjørehistorikken viser om datakvaliteten var på eller av for hver utførelse. Du kan også kjøre en suite on demand fra notatboken når som helst, uavhengig av oppdateringer.

Velg en hvilken som helst MLV-node i linjevisningen og åpne fanen Datakvalitet i detaljpanelet for å se resultatene av den kjøringen:

• Calculated Metrics — kolonnenivå-aggregatmetrikker beregnet av Deequ-analysatorene (for eksempel Completeness av region).
• Verifikasjonsresultater — en oppsummering av totalt antall kontroller, antall bestått og totalt mislykket, etterfulgt av hver begrensning og dens bestått/ikke bestått-status, slik at du kan se nøyaktig hvilken regel som bestod eller strøk.

Datakvalitetsfanen viser beregnede måleparametere og verifiseringsresultater per begrensning for den valgte MLV-kjøringen.

MLV-kjørehistorikken viser også en datakvalitetsstatus for hver kjøring — Suksess,Mislykket, eller ingen indikator når ingen suite kjørte — slik at du kan oppdage problemkjøringer med et blikk uten å åpne hver node.

Støttede sjekker og analysatorer

Fordi kolonnenivå-DQ drives av det åpne Deequ-biblioteket, kan du bruke alle Deequs sjekker og analysatorer inne i dine inntastingsfunksjoner. Valideringssjekker (brukt med @fmlv.validate) er Check begrensninger som gir et bestått/ikke bestått-resultat; analysatorer (brukt med @fmlv.analyze) beregner en metrikk uten bestått/ikke bestått. De vanligste er listet opp i følgende tabeller.

Valideringssjekker — @fmlv.validate

Analysatorer / måleparametere — @fmlv.analyze

Det åpne Deequ-biblioteket tilbyr disse sjekkene og analysatorene. For den komplette og autoritative listen, se Deequ-prosjektet på github.com/awslabs/deequ.

Summary

1. Opprett din MLV og noter det fullt kvalifiserte navnet.
2. I en separat datakvalitetscelle, registrer en suite med @fmlv.dq_suite(mlv_name=…).
3. Legg til sjekker — stack @fmlv.validate("<fn>", "FAIL"|"WARN") og/eller @fmlv.analyze("<fn>") skriv matchende oppføringsfunksjoner som returnerer PyDeequ DataFrames.
4. Kjør det — på forespørsel fra notatboken, eller slå på datakvalitetsbryteren når du lager en plan.
5. Gå gjennom resultater i fanen for kvalitet på slektslinje og i kjørehistorikken.

Beslektet innhold

• Datakvalitet i Materialized Lake Views — Microsoft Learn
• Oversikt over materialiserte Lake Views — Microsoft Learn
• DeeQu-dokumentasjon

Beslektet innhold

• Oppdater materialisert utsikt over innsjøen
• Power BI-rapporter for datakvalitet