Tietojen laatu muodostettuissa tallennustilanäkymissä

Medallion-arkkitehtuureissa datan laatu on valvottava jokaisessa vaiheessa. Huono tietojen laatu voi johtaa virheellisiin merkityksellisiin tietoihin ja operatiiviseen tehotkuuteen.

Tässä artikkelissa selitetään, miten datan laadun tarkistukset toteutetaan materialisoiduissa järvinäkymissä (MLV) Microsoft Fabricissa.

Tietojen laadun toteuttaminen

Rivitason rajoitteet validoivat jokaisen rivin erikseen (esimerkiksi "customerName ei ole null tällä rivillä"). Monet todelliset laatuongelmat ovat kuitenkin koko sarakkeen ominaisuuksia:

• Yksilöllisyys — pääavain tai liiketoimintaavain, jossa order_id ei ole kaksoiskappaleita.
• Täydellisyys — kriittinen sarakke, kuten , customer_id ei sisällä nollapisteitä.
• Tilastollinen muoto — sarakkeen keskiarvo, koko tai jakauma pysyy odotetulla alueella.

Näiden ongelmien havaitseminen virkistyshetkellä estää huonon datan leviämisen pronssi→hopea→kultaketjun läpi ja antaa auditoitavan laadukkaan signaalin jokaisella ajokerralla.

• Rivitason datan laaturajoitteet
• Saraketason datan laadun tarkistukset

Rivitason datan laaturajoitteet

Materialisoiduissa järvinäkymissä (MLV) Fabric-ohjelmassa ylläpidä datan laatu määrittämällä rajoituksia näkymiisi. Ilman tarkistuksia pienet dataongelmat voivat pidentää käsittelyaikaa tai vikata putken.

Kun rivi rikkoo rajoitteen, voit käyttää jotakin seuraavista toiminnoista:

• EPÄONNISTUMINEN: Estää MLV-päivityksen ensimmäisen rajoiterikkomuksen kohdalla. Tämä on oletuskäyttäytyminen, vaikka et määrittäisikään FAIL.

• DROP: Jatkaa käsittelyä ja poistaa tietueita, jotka rikkovat rajoitetta. Sukulinjanäkymä näyttää pudonneiden tietueiden määrän.

Määrittele datan laadun tarkistukset materialisoidussa järvinäkymässä

Kun luot materialisoidun järven näkymän, voit määritellä rajoitteita — datan laatusääntöjä, jotka vahvistavat jokaisen rivin päivityksen aikana. Rajoite on Boolen lauseke, jonka jokaisen rivin on täytettävä. Läpäisevät rivit kirjoitetaan tulostauluun. Rivejä, jotka epäonnistuvat, käsitellään on-violation -asetuksen mukaisesti: ne joko pudotetaan hiljaisesti tai aiheuttavat koko päivityksen epäonnistumisen.

Seuraavassa esimerkissä määritetään rajoite cust_blank, joka tarkistaa, onko customerName kenttä tyhjäarvo. Rajoite sulkee pois nollan customerName rivit käsittelystä.

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;

Järjestelmän sisäänrakennetut toiminnot

Sisäänrakennetut Spark/SQL-toiminnot, kuten UPPER(), LOWER(), TRIM(), COALESCE(), INITCAP() ja DATE_FORMAT() ovat täysin tuettuja kaikissa MLV-konteksteissa sekä CREATE- että LINEAGE-päivityksessä.

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:t – Määritelty ja rekisteröity samaan muistikirjaan

UDF:t, jotka on rekisteröity spark.udf.register():llä samassa muistikirjassa, tuetaan CREATE:lle kaikissa konteksteissa. LINEAGE-päivityksessä tuetaan vain PySpark-konteksteja, koska UDF-määritelmä suoritetaan osana aikataulutettua muistikirjan suoritusta.

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

Kolmannen osapuolen kirjastot - Pandas UDF:t

Kolmannen osapuolen kirjastot, kuten Pandas UDF:t, mahdollistavat datan laatusääntöjen toteuttamisen vektoroidulla prosessoinnilla. Ne mahdollistavat edistyneet validoinnit, kuten räätälöidyt liiketoimintalogiikan, tilastolliset tarkistukset tai kuvioiden tunnistuksen, joita ei ole mahdollista sisäänrakennetuilla funktioilla. Tämä auttaa rakentamaan skaalautuvia ja uudelleenkäytettäviä datan laaturajoitteita MLV:n luomisessa ja päivityksessä.

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

Mukautetut kirjastot – Python-pyörä (.whl)

Jar- tai wheel-tiedostoina pakatut funktiot voidaan asentaa Fabric-klusteriin (ympäristöasetusten kautta) ja käyttää MLV-määrittelyissä. CREATE ja LINEAGE REFRESH on tuettu PySpark-konteksteissa. Katso lisätietoja kohdasta Hallinnoi mukautettuja kirjastoja Fabric-ympäristöissä .

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

Fabric-käyttäjätietofunktiot

Fabric-käyttäjätietotoiminnot (UDF:t) määritellään ja hallitaan keskitetysti Fabric-työtilassa. Ne ovat saatavilla mihin tahansa kannettavaan tai putkeen ilman, että niitä tarvitsee rekisteröidä uudelleen per istunto, mikä tekee niistä ihanteellisia tuotanto-MLV-putkistoille. Tämä ominaisuus on tuettu vain PySpark-konteksteissa sekä CREATE- että LINEAGE-päivityksessä. Lue lisää User Data Functions -toiminnoista täältä. Lisätietoja löytyy Fabric User Data Functions -yleiskatsauksesta.

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

Saraketason datan laadun tarkistukset

Näin se toimii

Määrittele sarakketason datan laatutarkistukset (DQ) kolmella Python-koristelulla, jotka sijoitat muistikirjan soluun erillään MLV-määritelmästä:

Koska paketti kohdistuu MLV:hen nimeltä, datan laatu voidaan omistaa, ajaa ja ajoittaa riippumatta MLV-määritelmästä — ja voit suorittaa tarkistuksia ilman päivitystä.

Vaihe 1: Aseta suoritusaika

Määritä riippuvuudet Fabric-ajonaikaiselle versiollesi ennen kuin määrität sarakkeen tason datan laadun tarkistukset.

Fabric-ajonajat ennen versiota 2.0

Lataa PyDeequ 1.7.0 -pyörä ja lataa se mukautettuna kirjastona Fabric-ympäristöösi. Ohjeet löytyvät kohdasta Lataa mukautetut kirjastot.

Fabric runtime 2.0 tai uudempi

Lisää seuraavat Spark-ominaisuudet:

Voit lisätä nämä ominaisuudet joko ympäristöön tai muistikirjaan:

• Ympäristö: Lisää ominaisuudet Spark Compute>Spark -ominaisuuksiin, tallenna ja julkaise ympäristö. Ohjeita löytyy kohdasta Lisää Spark-konfiguraatiot mukautettuun ympäristöön.
• Muistikirja: Lisää conf ominaisuudet komennon %%configure objektiin ensimmäiseen koodisoluun ennen Spark-istunnon aloittamista. Ohjeita löytyy kohdasta Spark session configuration magic command.

Vaihe 2: Määrittele saraketasoiset tarkistukset

Tässä esimerkissä määritellään sarakkeen tason datan laadun tarkistukset materialisoidulle sales_db.daily_orders järvinäkymälle, määriteltynä mlv_name .@fmlv.dq_suite

Laita ohjelmistorekisteri ja sen syöttötoiminnot yhteen tietovarastoon. Seuraava esimerkki suorittaa kaksi tarkistusta yhdellä validointikierroksella — customer_id niiden on oltava valmiita (ei nollapisteitä) ja order_id yksilöllisiä:

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

Lisää mittareita epäonnistumatta juoksussa. Pino läpikäynti @fmlv.analyze laskeaksesi sarakkemittarit (koko, täydellisyys, keskiarvo, ...) havaittavuuden varmistamiseksi. Analyysitulokset tallennetaan aina eikä koskaan estä suoritusta — hyödyllistä silloin, kun validointisäännöt ovat vielä kehittymässä.

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

Hallitse lopputulos: EPÄONNISTUMINEN vs. VAROITUS

Toiminto päättää @fmlv.validate , mitä tapahtuu, kun tarkistus epäonnistuu:

• FAIL — tarkistuksen epäonnistuminen epäonnistuu datan laadun suorituksessa ja estää siitä riippuvaiset MLV:t. Käytä sitä koville sopimuksille, kuten ensisijaisille avaimille ja pakollisille kentille.
• WARN — vika kirjataan, mutta ajo onnistuu silti, eikä myöhemmät MLV :t esty. Käytä sitä pehmeisiin tai neuvonantaviin sääntöihin.

Vaihe 3: Ota käyttöön datan laadun tarkistukset aikataulussa

Valitse, suoritetaanko sarakkeen tarkistukset osana päivitysaikataulua. Create new schedule -kokemuksen LisäasetuksissaDatan laadun tarkistusten pudotusvalikossa on kaksi vaihtoehtoa:

• Tarkista jokaisen suorituksen jälkeen — datan laatupaketti käynnistyy automaattisesti jokaisen MLV-päivityksen jälkeen kyseisessä aikataulussa.
• Ei tarkistuksia — datan laatu ohitetaan kyseisessä aikataulussa; MLV päivittää edelleen rytmiään.

Datan laatu tarkistaaaikataulun Lisäasetuksissa — valitse Tarkista jokaisen ajon jälkeen tai Ei tarkistuksia.

Miten epäonnistunut tarkistus vaikuttaa putkeen

Lineage-näkymässä datan laadun tarkistus gold.daily_dq_reportusage määriteltiin ja FAIL epäonnistui. Tämän seurauksena sen alavirran riippuvainen , gold.monthly_dq_report…, ohitetaan automaattisesti, kun taas riippumattomat haarat (gold.expected_regions, gold.region_mlv_dq_rep…) valmistuvat normaalisti.

Datan FAIL laadun tarkistus pysäyttää ketjun — epäonnistuneen solmun alavirran riippuvainen ohitetaan, mutta itsenäiset haarat jatkuvat.

Teloitusmääräys. Kun sarjassa on molemmat tyypit, kaikki @fmlv.validate funktiot suoritetaan ensin (määrittelyjärjestyksessä), sitten kaikki @fmlv.analyze funktiot. Epäonnistunut tarkistus kirjataan ja suoritus jatkuu; kokonaistulos ratkaistaan kaikkien tarkistusten jälkeen — suoritus epäonnistuu vain, jos FAIL-toimintotarkistus epäonnistuu.

Vaihe 4: Katso ja analysoi datan laaturaportti

Suoritushistoria kertoo, oliko datan laatu päällä vai pois jokaisella suorituksella. Voit myös ajaa ohjelmistopaketin tilauksesta sen muistikirjasta milloin tahansa, riippumatta päivityksestä.

Valitse mikä tahansa MLV-solmu lineage-näkymästä ja avaa tietopaneelin Data quality -välilehti nähdäksesi kyseisen suorituksen tulokset:

• Lasketut mittarit — saraketasoiset aggregaattimittarit, jotka on laskettu Daqui-analysoijien toimesta (esimerkiksi Completeness ).region
• Varmennustulokset — yhteenveto kaikista tarkistuksista, läpäisyistä ja epäonnistuneista, joita seuraa jokainen rajoite ja sen läpäisy/hylätty -tila, jotta näet tarkalleen, mikä sääntö meni läpi tai epäonnistui.

Datan laatuvälilehti näyttää lasketut mittarit ja rajoitekohtaiset varmennustulokset valitulle MLV-suoritukselle.

MLV:n suoritushistoria näyttää myös datan laadun tilan jokaiselle ajolle — Onnistui, epäonnistunut tai ei indikaattoria, kun mikään sarja ei ajettu — joten ongelmasuoritukset voi havaita yhdellä silmäyksellä avaamatta jokaista solmua.

Tuetut tarkistukset ja analysaattorit

Koska sarakketason DQ toimii Amazon Deequlla, voit käyttää koko Daqui-kirjastoa sisäänpääsytoiminnoissasi. Validointitarkistukset (käytetään ) @fmlv.validateovat Check rajoitteita, jotka tuottavat läpäisy/epäonnistumis-tuloksen; analysaattorit (käytetään ) @fmlv.analyzelaskevat mittarin ilman läpäisty/epäonnistumista. Yleisimmät on lueteltu seuraavissa taulukoissa.

Validointitarkistukset — @fmlv.validate

Analysaattorit / mittarit — @fmlv.analyze

Avoimen lähdekoodin Amazon Deequ -kirjasto tarjoaa nämä tarkistukset ja analysaattorit. Täydellisen ja auktoritatiivisen listan löydät Daqui-projektista github.com/awslabs/deequ.

Summary

1. Luo MLV:si ja merkitse sen täysin pätevä nimi.
2. Erillisessä tiedonlaatusolussa rekisteröi sarja .@fmlv.dq_suite(mlv_name=…)
3. Lisää tarkistuksia — pinota @fmlv.validate("<fn>", "FAIL"|"WARN") ja/tai @fmlv.analyze("<fn>") kirjoita vastaavat syötetoiminnot, jotka palauttavat PyDeequ DataFrames.
4. Aja se — pyynnöstä kannettavasta tai laita datan laatu päälle aikataulua luodessasi.
5. Tarkastele tuloksia lineage Data quality -välilehdellä ja suoritushistoriassa.

Aiheeseen liittyvä sisältö

• Datan laatu Materialized Lake Views -näkymässä — Microsoft Learn
• Materialisoitujen järvien näkymien yleiskatsaus — Microsoft Learn
• DeeQu-dokumentaatio

Aiheeseen liittyvä sisältö

• Päivitä muodostetut näkymänäkymät järvelle
• Power BI raportoi datan laadusta