Implementación de comprobaciones de validación

Las canalizaciones de datos suelen encontrar registros con valores que faltan, identificadores duplicados o valores que se encuentran fuera de intervalos aceptables. Sin comprobaciones de validación, estos problemas de calidad de datos propagan hacia abajo, lo que provoca análisis incorrectos, informes con errores y decisiones empresariales poco confiables. Las comprobaciones de validación detectan estos problemas en el punto de ingesta de datos, lo que garantiza que solo los datos de calidad fluyen a través de la canalización.

En esta unidad, aprenderá a implementar comprobaciones de validación de nulabilidad, cardinalidad de datos y comprobación de intervalos mediante las expectativas de canalizaciones de Lakeflow y las restricciones de tabla de Delta Lake.

Descripción de los enfoques de validación en Azure Databricks

Azure Databricks proporciona dos mecanismos principales para implementar comprobaciones de validación: expectativas de canalización y restricciones de tabla. Cada enfoque ofrece diferentes escenarios y ofrece funcionalidades distintas.

Las expectativas de pipeline aplican validaciones durante las transformaciones de datos en las canalizaciones de Lakeflow. Las expectativas permiten generar una advertencia, eliminar registros no válidos o generar un error en la canalización cuando los datos infringen tus reglas. Este enfoque funciona bien para las tablas de streaming y las vistas materializadas en las que necesita un control de calidad en tiempo real.

Las restricciones de tabla aplican reglas directamente en tablas de Delta Lake. Las restricciones rechazan datos no válidos en tiempo de escritura, lo que impide que los registros incorrectos entren en las tablas. Este enfoque se adapta al procesamiento por lotes y escenarios que requieren garantías estrictas de integridad de datos.

Con estos enfoques, puede validar tres dimensiones clave de la calidad de los datos: nulabilidad (garantizando que existen valores necesarios), cardinalidad de datos (comprobación de la unicidad cuando se esperaba) y comprobación de intervalos (confirmar que los valores se encuentran dentro de límites aceptables).

Implementación de comprobaciones de nulabilidad

La validación de nulabilidad garantiza que las columnas necesarias contengan valores. Considere una tabla de clientes donde email y customer_id siempre deben tener valores. Puede implementar estas comprobaciones utilizando expectativas en su pipeline.

Con Python, agregue decoradores de expectativas a la definición de la tabla de streaming:

from pyspark import pipelines as dp

@dp.table
@dp.expect_or_drop("valid_email", "email IS NOT NULL")
@dp.expect_or_drop("valid_customer_id", "customer_id IS NOT NULL")
def customers():
    return spark.readStream.table("raw.customers")

La misma validación en SQL usa la CONSTRAINT cláusula :

CREATE OR REFRESH STREAMING TABLE customers(
    CONSTRAINT valid_email EXPECT (email IS NOT NULL) ON VIOLATION DROP ROW,
    CONSTRAINT valid_customer_id EXPECT (customer_id IS NOT NULL) ON VIOLATION DROP ROW
) AS SELECT * FROM STREAM(raw.customers);

En el caso de las tablas de Delta Lake fuera de las canalizaciones, use restricciones NOT NULL en la definición de tabla:

CREATE TABLE customers (
    customer_id INT NOT NULL,
    email STRING NOT NULL,
    first_name STRING,
    last_name STRING
);

También puede agregar restricciones de nulabilidad a las tablas existentes:

ALTER TABLE customers ALTER COLUMN email SET NOT NULL;

Validación de la cardinalidad de datos

La validación de cardinalidad garantiza que las columnas esperadas contengan valores únicos. Esta comprobación es esencial para las claves principales, los identificadores de transacción y otros campos en los que los duplicados indican problemas de calidad de los datos.

Las expectativas de canalización podrían validar la cardinalidad comprobando si hay condiciones que indiquen problemas de singularidad. Por ejemplo, puede comprobar que un número de seguro social aparece solo una vez por persona:

from pyspark.sql.window import Window
from pyspark.sql.functions import count

@dp.table
@dp.expect("unique_ssn_per_person", "ssn_count = 1")
def employees():
    df = spark.table("raw.employees")
    w = Window.partitionBy("ssn")
    return df.withColumn("ssn_count", count("*").over(w))

Para obtener comprobaciones de cardinalidad más completas, combine las expectativas con la lógica de agregación en la transformación:

CREATE OR REFRESH MATERIALIZED VIEW order_validation AS
SELECT 
    order_id,
    COUNT(*) as occurrence_count
FROM orders
GROUP BY order_id
HAVING COUNT(*) > 1;

Esta vista materializada identifica los identificadores de orden duplicados, lo que le permite investigar y resolver problemas de cardinalidad.

Delta Lake admite restricciones de clave principal que documentan la unicidad esperada, aunque estas restricciones son informativas y no se aplican:

CREATE TABLE orders (
    order_id INT NOT NULL,
    customer_id INT,
    order_date DATE,
    CONSTRAINT orders_pk PRIMARY KEY (order_id)
);

Aplica la comprobación de intervalos

La validación del intervalo confirma que los valores numéricos, date y otros valores están dentro de límites aceptables. Esta comprobación detecta errores de entrada de datos, errores del sistema y problemas de integración que generan valores fuera del intervalo.

Defina las expectativas de intervalo mediante operadores de comparación o la BETWEEN cláusula :

@dp.table
@dp.expect_or_fail("valid_age", "age BETWEEN 0 AND 150")
@dp.expect_or_fail("valid_salary", "salary >= 0")
@dp.expect_or_fail("valid_hire_date", "hire_date <= current_date()")
def employees():
    return spark.readStream.table("raw.employees")

En SQL, aplique las mismas comprobaciones de intervalo:

CREATE OR REFRESH STREAMING TABLE transactions(
    CONSTRAINT valid_amount EXPECT (amount > 0) ON VIOLATION DROP ROW,
    CONSTRAINT valid_date EXPECT (transaction_date <= current_date()) ON VIOLATION DROP ROW,
    CONSTRAINT valid_quantity EXPECT (quantity BETWEEN 1 AND 10000) ON VIOLATION DROP ROW
) AS SELECT * FROM STREAM(raw.transactions);

Para las tablas de Delta Lake, use restricciones CHECK para aplicar intervalos en tiempo de escritura:

ALTER TABLE employees ADD CONSTRAINT valid_age CHECK (age >= 18 AND age <= 120);
ALTER TABLE transactions ADD CONSTRAINT positive_amount CHECK (amount > 0);

Las comprobaciones de intervalos también pueden validar reglas de negocios que combinan varias condiciones:

@dp.expect("valid_discount", """
    discount_percent >= 0 
    AND discount_percent <= 100
    AND (discount_percent <= 50 OR customer_tier = 'PREMIUM')
""")

Esta expectativa garantiza que los descuentos permanezcan dentro de los límites y exijan que solo los clientes premium reciban descuentos superiores a 50%.

Elección de acciones para errores de validación

Cuando se produce un error en la validación, elija cómo responde la canalización. Cada acción se adapta a diferentes requisitos empresariales y niveles de importancia de los datos.

Use la acción de advertencia cuando necesite visibilidad de los problemas de calidad de los datos sin bloquear el flujo de datos:

@dp.expect("has_phone", "phone_number IS NOT NULL")

Usa drop cuando los registros no válidos deban filtrarse silenciosamente:

@dp.expect_or_drop("complete_address", "street IS NOT NULL AND city IS NOT NULL")

Usa fallo cuando la integridad de los datos es crítica y los registros no válidos son inaceptables.

@dp.expect_or_fail("valid_account_balance", "balance >= 0")

Puede ver las métricas de expectativas en la interfaz de usuario de canalización seleccionando un conjunto de datos con expectativas y abriendo la pestaña Calidad de los datos . Estas métricas le ayudan a supervisar las tasas de aprobación de validación e identificar problemas sistemáticos de calidad de los datos.

Ahora que comprende cómo implementar comprobaciones de validación, puede asegurarse de que las canalizaciones de datos mantienen estándares de calidad para los intervalos de nulabilidad, cardinalidad y valor.