Qualidade dos dados em exibições de lago materializadas

Em arquiteturas de medalhão, você deve impor a qualidade dos dados em todos os estágios. A baixa qualidade dos dados pode levar a insights incorretos e ineficiências operacionais.

Este artigo explica como implementar verificações de qualidade de dados nas visões materializadas do lago de dados (MLVs) no Microsoft Fabric.

Implementar a qualidade dos dados

Restrições em nível de linha validam cada linha isoladamente (por exemplo, "customerName não é NULL nesta linha"). Muitos problemas reais de qualidade, no entanto, são propriedades de uma coluna como um todo:

• Unicidade — uma chave primária ou de negócios, como order_id, não possui duplicatas.
• Completude — uma coluna crítica como customer_id não contém nulos.
• Forma estatística — a média, tamanho ou distribuição de uma coluna permanece dentro de uma faixa esperada.

Detectar esses problemas no ponto de atualização impede que dados ruins se propaguem a jusante pela sua cadeia bronze→prata→ouro e te dá um sinal de qualidade auditável a cada execução.

• Restrições de qualidade de dados em nível de linha
• Verificações de qualidade de dados em nível de coluna

Restrições de qualidade de dados em nível de linha

Nas vistas de lago materializadas (MLVs) no Fabric, mantenha a qualidade dos dados definindo restrições para suas views. Sem verificações explícitas, pequenos problemas de dados podem aumentar o tempo de processamento ou falhar no pipeline.

Quando uma linha viola uma restrição, você pode usar uma destas ações:

• FAIL: interrompe a atualização do MLV na primeira violação de restrição. Esse é o comportamento padrão, mesmo quando você não especifica FAIL.

• DROP: continua o processamento e remove registros que violam a restrição. A visualização de linhagem mostra a contagem de registros descartados.

Definir verificações de qualidade de dados em uma exibição de lago materializada

Ao criar uma exibição de lago materializada, você pode definir restrições – regras de qualidade de dados que validam cada linha durante uma atualização. Uma restrição é uma expressão booliana que cada linha deve satisfazer. As linhas que passam são gravadas na tabela de saída. As linhas que falham são tratadas de acordo com a configuração de tratamento de violações: elas são descartadas silenciosamente ou fazem com que toda a atualização falhe.

O exemplo a seguir define a restrição cust_blank, que verifica se o customerName campo não é nulo. A restrição exclui linhas com um nulo customerName do processamento.

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;

Funções integradas ao sistema

Funções internas do Spark/SQL, como UPPER(), LOWER(), TRIM(), COALESCE(), INITCAP() e DATE_FORMAT() têm suporte total em todos os contextos MLV para atualizações de 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)

UDFs – Definidos e registrados no mesmo notebook

UDFs registrados com spark.udf.register() no mesmo notebook são compatíveis com CREATE em todos os contextos. Para a atualização LINEAGE, há suporte apenas para contextos do PySpark porque a definição de UDF é executada como parte da execução agendada do 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")

Bibliotecas de terceiros - Pandas UDFs

Bibliotecas de terceiros, como UDFs do Pandas, permitem que regras de qualidade de dados sejam implementadas com processamento vetorizado. Eles permitem validações avançadas, como lógica de negócios personalizada, verificações estatísticas ou detecção de padrões que não são possíveis com funções internas. Isso ajuda a criar restrições de qualidade de dados escalonáveis e reutilizáveis durante a criação e atualização do 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")

Bibliotecas personalizadas – wheel do Python (.whl)

As funções empacotadas como arquivos jar ou arquivos wheel podem ser instaladas no Fabric cluster (por meio de configurações de ambiente) e usadas em definições de MLV. CREATE e LINEAGE REFRESH têm suporte para contextos do PySpark. Consulte Gerenciar bibliotecas personalizadas em ambientes do Fabric para obter mais detalhes.

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

Funções de dados do usuário do Fabric

As UDFs (Funções de Dados do Usuário) do Fabric são definidas centralmente e gerenciadas no workspace do Fabric. Eles estão disponíveis para qualquer notebook ou pipeline sem a necessidade de serem registrados novamente por sessão, tornando-os ideais para pipelines de Machine Learning e Visualização (MLV) de produção. Essa funcionalidade só tem suporte em contextos PySpark para atualização de CREATE e de LINEAGE. Saiba mais sobre as funções de dados do usuário aqui. Para obter mais informações, consulte a visão geral do 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)

Verificações de qualidade de dados em nível de coluna

Como funciona

Defina verificações de qualidade de dados em nível de coluna (DQ) com três decoradores Python que você coloca em uma célula de notebook, separadas da definição de MLV:

Como a suíte tem como alvo um MLV pelo nome, a qualidade dos dados pode ter um responsável definido, ser executada e agendada independentemente da definição do MLV — e você pode executar verificações sem acionar uma atualização.

Passo 1: Configurar o runtime

Configure as dependências para a versão do runtime do Fabric antes de definir verificações de qualidade de dados em nível de coluna.

runtimes do Fabric anteriores à versão 2.0

Baixe a PyDeequ 1.7.0 wheel e faça o upload como uma biblioteca personalizada no seu ambiente Fabric. Para obter instruções, consulte Fazer upload de bibliotecas personalizadas.

Fabric runtime 2.0 ou posterior

Adicione as seguintes propriedades do Spark:

Você pode adicionar essas propriedades tanto em um ambiente quanto em um notebook:

• Ambiente: Adicione as propriedades em Spark Compute>Propriedades Spark, depois salve e publique o ambiente. Para instruções, veja Adicionar configurações do Spark a um ambiente personalizado.
• Caderno: Adicione as propriedades no objeto conf de um comando %%configure na primeira célula de código, antes de iniciar a sessão do Spark. Para instruções, veja o comando mágico de configuração da sessão Spark.

Passo 2: Defina verificações em nível de coluna

Neste exemplo, defina verificações de qualidade de dados em nível de coluna para a sales_db.daily_orders visão materializada do lake, especificadas por mlv_name em @fmlv.dq_suite.

Coloque o registro da suíte e suas funções de inicialização em uma única célula de qualidade de dados. O exemplo a seguir executa duas verificações em uma única passagem de validação — customer_id deve estar completo (sem nulos) e order_id não deve se repetir:

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

Adicione métricas sem falhar na execução. Empilhe um @fmlv.analyze pass para calcular métricas de coluna (tamanho, completude, média, ...) para observabilidade. Os resultados de análise são sempre registrados e nunca bloqueiam uma execução — útil enquanto as regras de validação ainda estão evoluindo.

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

Controle o resultado: FAIL vs. WARN

A ação em @fmlv.validate decide o que acontece quando uma verificação falha:

• FAIL — a falha na verificação faz com que a execução de qualidade dos dados falhe e bloqueia os MLVs downstream que dependem dela. Use-o para restrições rígidas, como chaves primárias e campos obrigatórios.
• WARN — a falha é registrada, mas a execução ainda é bem-sucedida e os MLVs a jusante não são bloqueados. Use para regras suaves ou de aviso.

Passo 3: Ative as verificações de qualidade dos dados em um cronograma

Escolha se as verificações em nível de coluna são executadas como parte de um cronograma de atualização. Nas configurações avançadas da experiência Criar novo cronograma , o menu suspenso de verificações de qualidade de dados oferece duas opções:

• Verifique após cada execução — o conjunto de qualidade de dados roda automaticamente após cada atualização MLV nesse cronograma ser concluído.
• Sem verificações — a verificação da qualidade dos dados é ignorada nesse cronograma; o MLV ainda atualiza em sua cadência.

A configuração de verificação de qualidade de dados nas configurações avançadas de um agendamento — escolha Verificar após cada execução ou Sem verificações.

Como uma verificação com falha afeta o pipeline

Na visão de linhagem, a verificação de qualidade dos dados em gold.daily_dq_reportusage foi definida com FAIL e falhou. Como resultado, seu dependente a jusante, gold.monthly_dq_report…, é automaticamente ignorado, enquanto os ramos independentes (gold.expected_regions, gold.region_mlv_dq_rep…) são concluídos normalmente.

Uma FAIL verificação de qualidade dos dados interrompe a cadeia — o dependente a jusante do nó falhado é pulado, mas as ramificações independentes continuam.

Ordem de execução. Quando um conjunto tem ambos os tipos, todas @fmlv.validate as funções executam primeiro (em ordem de definição), depois todas @fmlv.analyze as funções. Uma verificação com falha é registrada e a execução continua; o resultado geral é decidido após todas as verificações serem concluídas — a execução falha apenas se uma verificação de ação FAIL falhou.

Passo 4: Visualize e analise o relatório de qualidade dos dados

O histórico de execução reflete se a qualidade dos dados estava ligada ou desligada em cada execução. Você também pode rodar um conjunto sob demanda a partir do seu notebook a qualquer momento, independentemente de qualquer atualização.

Selecione qualquer nó MLV na visualização de linhagem e abra a aba Qualidade de Dados no painel de detalhes para ver os resultados dessa execução:

• Métricas Calculadas — as métricas agregadas em nível de coluna calculadas pelos analisadores Deequ (por exemplo, Completeness de region).
• Resultados de Verificação — um resumo do total de verificações, total de aprovados e total de falhas, seguido por cada restrição e seu Status de aprovado/reprovado, para que você possa ver exatamente qual regra passou ou falhou.

A aba Qualidade dos Dados mostra métricas calculadas e resultados de verificação por restrição para a execução MLV selecionada.

O histórico de execuções do MLV também mostra um status de qualidade de dados para cada execução — Sucesso, Falhado ou nenhum indicador quando nenhuma suíte rodou — então você pode identificar as execuções de problemas rapidamente sem abrir cada nó.

Verificações e analisadores suportados

Como o DQ em nível de coluna é alimentado pelo Amazon Deequ, você pode usar toda a biblioteca Deequ dentro das suas funções de entrada. Verificações de validação (usadas com @fmlv.validate) são Check restrições que produzem um resultado de aprovação/reprovação; analisadores (usados com @fmlv.analyze) calculam uma métrica sem aprovação/reprovação. Os mais comuns estão listados nas tabelas a seguir.

Verificações de validação — @fmlv.validate

Analisadores / métricas — @fmlv.analyze

A biblioteca open source Amazon Deequ fornece essas verificações e analisadores. Para a lista completa e autoritária, veja o projeto Deequ em github.com/awslabs/deequ.

Summary

1. Crie seu MLV e registre seu nome qualificado completo (FQN).
2. Em uma célula separada de qualidade de dados, registre uma suite com @fmlv.dq_suite(mlv_name=…).
3. Adicionar verificações — empilhar @fmlv.validate("<fn>", "FAIL"|"WARN") e/ou @fmlv.analyze("<fn>") e escrever as funções de entrada correspondentes que retornam DataFrames do PyDeequ.
4. Execute-o — sob demanda do notebook, ou ative a opção de qualidade de dados ao criar um cronograma.
5. Revise os resultados na aba Qualidade dos dados de linhagem e no histórico de execuções.

Conteúdo relacionado

• Qualidade dos dados nas Exibições de Lake Materializadas — Microsoft Learn
• Visão geral do Materialized Lake Views — Microsoft Learn
• Documentação do DeeQu

Conteúdo relacionado

• Atualizar vistas materializadas do lago
• Relatórios do Power BI sobre qualidade de dados