Qualidade dos dados em vistas materializadas do lago

Nas arquiteturas medallion, deve garantir a qualidade dos dados em todas as fases. A má qualidade dos dados pode levar a insights incorretos e ineficiências operacionais.

Este artigo explica como implementar verificações de qualidade de dados em visualizações de lagos materializadas (MLVs) no Microsoft Fabric.

Implementar a qualidade dos dados

Restrições ao nível da 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ócio, como order_id, não tem duplicados.
• 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 mantém-se dentro de um intervalo esperado.

Detetar estes problemas no ponto de atualização impede que dados errados se propaguem a jusante através da sua cadeia bronze→silver→gold, e fornece um sinal de qualidade auditável em cada execução.

• Restrições de qualidade dos dados ao nível da linha
• Verificações de qualidade de dados ao nível das colunas

Restrições de qualidade dos dados ao nível da linha

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

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

• FALHA: Impede a atualização do MLV na primeira violação de restrição. Este é o comportamento padrão, mesmo quando não especificas FAIL.

• DROP: Continua a processar e remove registos que violem a restrição. A vista de linhagem mostra a contagem de registos descartados.

Defina verificações de qualidade dos dados numa vista materializada do lago

Quando cria uma vista de lago materializada, pode definir restrições — regras de qualidade de dados que validam cada linha durante uma atualização. Uma restrição é uma expressão booleana que toda linha deve satisfazer. As linhas que passam são escritas na tabela de saída. As linhas que falham são tratadas de acordo com a configuração de violação: ou são descartadas sem aviso ou fazem com que toda a operação de 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 no sistema

Funções Spark/SQL incorporadas como UPPER(), LOWER(), TRIM(), COALESCE(), INITCAP(e DATE_FORMAT() são totalmente suportadas em todos os contextos MLV tanto para CREATE como para 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)

UDFs – Definidos e registados no mesmo notebook

UDFs registados com spark.udf.register() no mesmo notebook são suportados para CREATE em todos os contextos. Para a atualização LINEAGE, apenas os contextos PySpark são suportados porque a definição 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 os Pandas UDFs permitem a implementação de regras de qualidade de dados com processamento vetorizado. Permitem validações avançadas como lógica de negócio personalizada, verificações estatísticas ou deteção de padrões que não são possíveis com funções integradas. Isto ajuda a construir restrições de qualidade de dados escaláveis e reutilizáveis durante a criação e atualização de 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 Python (.whl)

Funções empacotadas como ficheiros jar ou ficheiros de roda podem ser instaladas no cluster Fabric (através das definições de Ambiente) e usadas em definições MLV. CREATE e LINEAGE REFRESH são suportados para contextos PySpark. Consulte Gerir bibliotecas personalizadas em ambientes Fabric para 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 de Utilizador do Fabric

As Funções de Dados de Utilizador do Fabric (UDFs) são definidas e geridas centralmente no espaço de trabalho do Fabric. Estão disponíveis para qualquer notebook ou pipeline sem necessidade de um novo registo por sessão, o que os torna ideais para pipelines MLV em produção. Esta funcionalidade é suportada apenas em contextos PySpark tanto para CREATE como para LINEAGE refresh. Saiba mais sobre as Funções de Dados do Utilizador aqui. Para mais informações, consulte a visão geral das Funções de Dados do Utilizador do 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)

Verificações de qualidade de dados ao nível das colunas

Como funciona

Defina verificações de qualidade de dados ao nível das colunas (DQ) com três decoradores Python que coloca numa célula de caderno, separada da definição MLV:

Como a suíte tem como alvo um MLV pelo nome, a gestão da qualidade dos dados pode ser feita, executada e agendada independentemente da definição do MLV — e pode executar verificações sem desencadear uma atualização.

Passo 1: Configurar o runtime

Configure as dependências para a sua versão de runtime Fabric antes de definir verificações de qualidade dos dados ao nível das colunas.

Tempos de execução Fabric anteriores à versão 2.0

Descarregue o wheel PyDeequ 1.7.0 e carregue-o como uma biblioteca personalizada no seu ambiente Fabric. Para instruções, consulte Carregar bibliotecas personalizadas.

Fabric runtime 2.0 ou posterior

Adicione as seguintes propriedades do Spark:

Pode adicionar estas propriedades num ambiente ou num caderno:

• Ambiente: Adicione as propriedades em Spark Compute>Propriedades do Spark, depois guarde e publique o ambiente. Para instruções, consulte Adicionar configurações de Spark a um ambiente personalizado.
• Caderno: Adicione as propriedades no conf objeto de um %%configure comando 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: Definir verificações ao nível das colunas

Neste exemplo, defina verificações de qualidade de dados ao nível das colunas para a sales_db.daily_orders vista lake materializada, especificadas por mlv_name em @fmlv.dq_suite.

Coloque o registo da suite e as suas funções de ponto de entrada numa única célula de qualidade de dados. O exemplo seguinte executa duas verificações numa única passagem de validação — customer_id deve ser completo (sem nulos) e order_id deve ser único:

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

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

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

Controlar o resultado: FAIL vs. WARN

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

• FAIL — a falha na verificação faz falhar a execução da qualidade dos dados e bloqueia os MLVs a jusante que dela dependem. Use-o para restrições rígidas, como chaves primárias e campos obrigatórios.
• WARN — a falha é registada, mas a execução continua a ter sucesso e os MLVs a jusante não são bloqueados. Usa-o para regras suaves ou informativas.

Passo 3: Ativar verificações de qualidade dos dados num agendamento

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

• Verifique após cada execução — a suíte de qualidade de dados corre automaticamente após cada atualização MLV nesse calendário terminar.
• Sem verificações — a qualidade dos dados é ignorada nesse agendamento; o MLV continua a ser atualizado de acordo com a sua cadência.

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

Como um teste falhado afeta o pipeline

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

Uma FAIL verificação de qualidade dos dados interrompe a cadeia — o dependente a jusante do nó falhado é ignorado, mas os ramos independentes continuam.

Ordem de execução. Quando uma suite tem ambos os tipos, todas @fmlv.validate as funções executam primeiro (por ordem de definição), depois todas @fmlv.analyze as funções. Uma verificação falhada é registada e a execução continua; o resultado global é decidido após a conclusão de todas as verificações — a sequência falha apenas se uma verificação de FAIL-ação falhar.

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

O histórico de execuções reflete se a qualidade dos dados estava ligada ou desligada em cada execução. Também pode executar uma suite a pedido a partir do seu notebook a qualquer momento, independentemente de qualquer atualização.

Selecione qualquer nó MLV na vista de linhagem e abra o separador Qualidade de Dados do painel de detalhes para ver os resultados dessa execução:

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

O separador 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 o estado da qualidade dos dados para cada execução — Sucesso, Falhado, ou nenhum indicador quando nenhuma suíte foi executada — para que possa identificar as execuções problemáticas de imediato sem abrir cada nó.

Verificações e analisadores suportados

Como o DQ ao nível da coluna é suportado por Amazon Deequ, pode usar toda a biblioteca Deequ nas suas funções de entrada. Verificações de validação (usadas com @fmlv.validate) são Check restrições que produzem um resultado de sucesso/falha; analisadores (usados com @fmlv.analyze) calculam uma métrica sem resultado de sucesso/falha. Os mais comuns estão listados nas tabelas seguintes.

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

Analisadores / métricas — @fmlv.analyze

A biblioteca open source Amazon Deequ fornece estas verificações e analisadores. Para a lista completa e autoritativa, consulte o projeto Deequ em github.com/awslabs/deequ.

Summary

1. Crie o seu MLV e anote o nome totalmente qualificado.
2. Numa célula separada de qualidade de dados, regista um conjunto 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 devolvem DataFrames PyDeequ.
4. Execute-o — a pedido a partir do notebook, ou ativa a opção de qualidade dos dados ao criar um agendamento.
5. Consulte os resultados no separador Qualidade dos dados da linhagem e no histórico de execuções.

Conteúdo relacionado

• Qualidade dos dados em Materialized Lake Views — Microsoft Learn
• Descrição geral de Visualizações Materializadas do Lago — Microsoft Learn
• Documentação da DeeQu

Conteúdo relacionado

• Atualize as vistas materializadas do lago
• Relatórios do Power BI sobre a qualidade dos dados