Calidad de los datos en vistas materializadas del lago

En las arquitecturas de medallón, debe asegurar la calidad de los datos en cada fase. La mala calidad de los datos puede dar lugar a conclusiones incorrectas e ineficacias operativas.

En este artículo se explica cómo implementar comprobaciones de calidad de datos en vistas materializadas de lago (MLV) en Microsoft Fabric.

Implementación de la calidad de los datos

Las restricciones a nivel de fila validan cada fila de forma aislada (por ejemplo, "customerName no es nulo en esta fila"). Sin embargo, muchos problemas reales de calidad son las propiedades de una columna en su conjunto:

• Unicidad — una clave primaria o de negocio, como order_id, no tiene duplicados.
• Completitud — una columna crítica como customer_id no contiene nulos.
• Forma estadística — la media, tamaño o distribución de una columna se mantiene dentro de un rango esperado.

Detectar estos problemas en el punto de actualización evita que los datos erróneos se propaguen aguas abajo a través de tu cadena bronce→plata→oro, y te da una señal de calidad auditable en cada pasada.

• Restricciones de calidad de datos a nivel de fila
• Comprobaciones de calidad de datos a nivel de columna

Restricciones de calidad de datos a nivel de fila

En las vistas de lago materializadas (MLV) en Fabric, mantén la calidad de los datos definiendo restricciones sobre tus vistas. Sin comprobaciones explícitas, los problemas de datos menores pueden aumentar el tiempo de procesamiento o producir errores en la canalización.

Cuando una fila infringe una restricción, puede usar una de estas acciones:

• FAIL: detiene la actualización de MLV en la primera violación de restricciones. Este es el comportamiento predeterminado, incluso cuando no se especifica FAIL.

• DROP: continúa procesando y quita los registros que infringen la restricción. La vista de linaje muestra el recuento de registros descartados.

Definición de comprobaciones de calidad de datos en una vista de lago materializada

Al crear una vista de lago materializada, puede definir restricciones: reglas de calidad de datos que validan cada fila durante una actualización. Una restricción es una expresión booleana que cada fila debe satisfacer. Las filas que pasan se registran en la tabla de salida. Las filas que producen un error se controlan según la configuración ante infracciones: se eliminan silenciosamente o provocan el fallo de toda la actualización.

En el ejemplo siguiente se define la restricción cust_blank, que comprueba si el customerName campo no es NULL. La restricción excluye las filas con un valor NULL customerName del procesamiento.

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;

Funciones integradas del sistema

Las funciones de Spark/SQL integradas, como UPPER(), LOWER(), TRIM(), COALESCE(), INITCAP() y DATE_FORMAT() son totalmente compatibles con todos los contextos de MLV para la actualización CREATE y 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 y registrados en el mismo cuaderno

Las UDF registradas con spark.udf.register() en el mismo cuaderno se admiten para CREATE en todos los contextos. Para la actualización de LINEAGE, solo se admiten los contextos de PySpark porque la definición de las UDF se ejecuta como parte de la ejecución programada del cuaderno.

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 terceros - Pandas UDFs

Las bibliotecas de terceros, como las UDF de Pandas, permiten implementar reglas de calidad de datos con procesamiento vectorizado. Habilitan validaciones avanzadas, como lógica de negocios personalizada, comprobaciones estadísticas o detección de patrones que no son posibles con funciones integradas. Esto ayuda a crear restricciones de calidad de datos escalables y reutilizables durante la creación y actualización 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 de Python (.whl)

Las funciones empaquetadas como archivos jar o archivos wheel se pueden instalar en el clúster de Fabric (a través de la configuración del entorno) y se usan en las definiciones de MLV. CREATE y LINEAGE REFRESH se admiten en contextos de PySpark. Consulte Administración de bibliotecas personalizadas en entornos de Fabric para obtener más detalles.

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

Funciones de datos de usuario de Fabric

Las funciones de datos de usuario (UDF) de Fabric se definen y administran de forma centralizada en el área de trabajo de Fabric. Están disponibles para cualquier cuaderno o canalización sin necesidad de volver a registrarse por sesión, por lo que son ideales para canalizaciones de MLV de producción. Esta característica solo se admite en contextos de PySpark para actualizar tanto CREATE como LINEAGE. Obtenga más información sobre las funciones de datos de usuario aquí. Para obtener más información, consulte Información general sobre Las funciones de datos de usuario de 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)

Comprobaciones de calidad de datos a nivel de columna

Cómo funciona

Defina comprobaciones de calidad de datos a nivel de columna (DQ) con tres decoradores de Python que coloque en una celda de cuaderno, separadas de la definición MLV:

Como la suite apunta a un MLV por nombre, la calidad de los datos puede gestionarse, ejecutarse y programarse independientemente de la definición de MLV, y puedes ejecutar comprobaciones sin activar una actualización.

Paso 1: Configurar el entorno de ejecución

Configura las dependencias de tu versión de ejecución de Fabric antes de definir controles de calidad de datos a nivel de columna.

Fabric runtimes anteriores a la versión 2.0

Descarga el paquete wheel de PyDeequ 1.7.0 y súbelo como una biblioteca personalizada en tu entorno de Fabric. Para obtener instrucciones, consulta Cargar bibliotecas personalizadas.

Runtime de Fabric 2.0 o posterior

Añade las siguientes propiedades de Spark:

Puedes añadir estas propiedades tanto en un entorno como en un cuaderno:

• Entorno: Añade las propiedades bajo Spark Compute>Propiedades de Spark, luego guarda y publica el entorno. Para las instrucciones, consulta Añadir configuraciones de Spark a un entorno personalizado.
• Cuaderno: Añadir las propiedades en el conf objeto de un %%configure comando en la primera celda de código, antes de iniciar la sesión de Spark. Para las instrucciones, consulta el comando mágico de configuración de sesión de Spark.

Paso 2: Definir comprobaciones a nivel de columna

En este ejemplo, definamos comprobaciones de calidad de datos a nivel de columna para la sales_db.daily_orders vista materializada del lago, especificadas por mlv_name en @fmlv.dq_suite.

Coloca el registro del conjunto y sus funciones de entrada en una sola celda de calidad de datos. El siguiente ejemplo ejecuta dos comprobaciones en un solo pase de validación — customer_id debe estar completo (sin nulos) y order_id debe 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)

Añade métricas sin provocar un fallo en la ejecución. Añade un @fmlv.analyze pass para calcular métricas de columna (tamaño, completitud, media, ...) para observabilidad. Los resultados del análisis siempre se registran y nunca bloquean una ejecución — útil mientras las reglas de validación aún están evolucionando.

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

Controla el resultado: FAIL vs. WARN

La acción sobre @fmlv.validate decide qué ocurre cuando una prueba falla:

• FAIL — el fallo de la comprobación provoca un fallo en la ejecución de calidad de datos y bloquea las MLV posteriores que dependen de él. Úsalo para restricciones estrictas como claves primarias y campos obligatorios.
• WARN — el fallo se registra pero la ejecución se completa correctamente y las MLV posteriores no se bloquean. Úsalo para reglas blandas o no obligatorias.

Paso 3: Habilitar las comprobaciones de calidad de los datos en una programación

Elige si las comprobaciones a nivel de columna se ejecutan como parte de un calendario de actualización. En la configuración avanzada de la experiencia Crear nueva programación, el desplegable de comprobaciones de calidad de datos ofrece dos opciones:

• Comprueba después de cada ejecución — la suite de calidad de datos se ejecuta automáticamente después de que finalice cada actualización de MLV de esa programación.
• Sin comprobaciones — la calidad de los datos se omite en ese calendario; el MLV se sigue actualizando según su cadencia.

La configuración de comprobaciones de calidad de datos en la configuración avanzada de una programación — elige Comprobar después de cada ejecución o No comprobar.

Cómo afecta una comprobación fallida a la canalización

En la vista de linaje, la comprobación de calidad de datos en gold.daily_dq_reportusage se definió con FAIL y falló. Como resultado, su elemento dependiente aguas abajo, gold.monthly_dq_report…, se omite automáticamente, mientras que las ramas independientes (gold.expected_regions, gold.region_mlv_dq_rep…) se completan normalmente.

Una FAIL comprobación de calidad de datos detiene la cadena — se omite el dependiente aguas abajo del nodo fallido, pero continúan las ramas independientes.

Orden de ejecución. Cuando una suite tiene ambos tipos, todas @fmlv.validate las funciones se ejecutan primero (en orden de definición) y luego todas @fmlv.analyze las funciones. Se registra una comprobación fallida y la ejecución continúa; el resultado general se decide tras completar todas las comprobaciones — la ejecución falla solo si una comprobación de acción FAIL falló.

Paso 4: Consulta y analiza el informe de calidad de los datos

El historial de ejecuciones refleja si la calidad de los datos estaba activada o desactivada en cada ejecución. También puedes ejecutar una suite bajo demanda desde su cuaderno en cualquier momento, independientemente de cualquier actualización.

Selecciona cualquier nodo MLV en la vista de linaje y abre la pestaña de calidad de datos del panel de detalles para ver los resultados de esa ejecución:

• Métricas calculadas — las métricas agregadas a nivel de columna calculadas por los analizadores Deequ (por ejemplo, Completeness de region).
• Resultados de verificación — un resumen de las comprobaciones totales, el total aprobado y el total fallido, seguido de cada restricción y su estado de aprobado/fallado, para que puedas ver exactamente qué regla ha pasado o fallado.

La pestaña de calidad de datos muestra métricas calculadas y resultados de verificación por restricción para la ejecución MLV seleccionada.

El historial de las ejecuciones de MLV también muestra un estado de calidad de datos para cada ejecución — Éxito, Fallido o sin indicador cuando ninguna suite se ejecutó — por lo que puedes detectar las ejecuciones de problemas de un vistazo sin abrir cada nodo.

Comprobaciones y analizadores compatibles

Como DQ a nivel de columna funciona con Amazon Deequ, puedes usar toda la biblioteca Deequ dentro de tus funciones de entrada. Las comprobaciones de validación (usadas con @fmlv.validate) son Check restricciones que producen un resultado de aprobado/fallo; los analizadores (usados con @fmlv.analyze) calculan una métrica sin aprobado/fallo. Los más comunes se enumeran en las siguientes tablas.

Comprobaciones de validación — @fmlv.validate

Analizadores / métricas — @fmlv.analyze

La biblioteca de código abierto Amazon Deequ proporciona estas comprobaciones y analizadores. Para la lista completa y autorizada, véase el proyecto Deequ en github.com/awslabs/deequ.

Summary

1. Crea tu MLV y apunta su nombre totalmente cualificado.
2. En una celda de calidad de datos separada, registre una suite con @fmlv.dq_suite(mlv_name=…).
3. Añadir comprobaciones — apila @fmlv.validate("<fn>", "FAIL"|"WARN") y/o @fmlv.analyze("<fn>") y escribe las funciones de entrada correspondientes que devuelven DataFrames de PyDeequ.
4. Ejecutarlo — bajo demanda desde el cuaderno, o activar el interruptor de calidad de datos al crear una programación.
5. Revisa los resultados en la pestaña de calidad de datos de linaje y en el historial de ejecuciones.

Contenido relacionado

• Calidad de datos en Materialized Lake Views — Microsoft Learn
• Resumen de Materialized Lake Views — Microsoft Learn
• Documentación de DeeQu

Contenido relacionado

• Actualizar vistas materializadas del lago
• Informes de Power BI para la calidad de los datos