Auditoría y Validación de Calidad de Datos a Gran Escala con Great Expectations y PySpark

🚀 Introducción a la Calidad de Datos en Entornos Big Data

En el ecosistema de la ciencia de datos moderna, el volumen de información que procesamos crece exponencialmente. Sin embargo, un gran volumen sin controles adecuados solo conduce a tomar decisiones erróneas a gran velocidad. El fenómeno conocido como Garbage In, Garbage Out se magnifica en los entornos de Big Data, donde los errores en las fuentes de datos pueden corromper modelos de Machine Learning enteros o tableros ejecutivos.

Tradicionalmente, la validación de datos se realizaba mediante scripts ad-hoc escritos en SQL o Python puro que resultaban difíciles de mantener y escalar. Hoy en día, herramientas como Great Expectations (GX) combinadas con motores distribuidos como PySpark nos permiten definir contratos de datos robustos, reutilizables y nativos para la nube.

Al finalizar este tutorial, serás capaz de:

• Diseñar expectativas de validación personalizadas para datasets masivos.
• Configurar un entorno de ejecución distribuido usando PySpark.
• Automatizar reportes de calidad que alerten a tu equipo ante cualquier anomalía.

🛠️ Requisitos Previos y Configuración del Entorno

Antes de escribir código, asegurémonos de tener las herramientas necesarias instaladas en nuestro entorno de desarrollo. Trabajaremos con Python 3.10 o superior, Apache Spark y la librería de Great Expectations.

Dependencias del Sistema

Necesitarás instalar las siguientes librerías mediante tu gestor de paquetes favorito. Asegúrate de tener Java (JDK 8 o 11) instalado en tu sistema, ya que es un requisito fundamental para que Apache Spark funcione correctamente.

pip install pyspark great-expectations pandas

📐 Arquitectura del Sistema de Validación

Para entender cómo interactúan Great Expectations y PySpark bajo el capó, analicemos el siguiente diagrama de flujo que representa el ciclo de vida de una validación de datos distribuidos.

La gran ventaja de esta arquitectura es que los datos nunca abandonan el clúster de Spark. Great Expectations traduce las expectativas de negocio a operaciones optimizadas de PySpark, evaluando millones de filas en segundos sin causar cuellos de botella en la memoria de una sola máquina.

💻 Implementación Práctica: Validando un Dataset Masivo

Imagina que trabajas para una plataforma de comercio electrónico global y recibes diariamente transacciones en formato Parquet que superan los 50 millones de registros. Vamos a construir un script para validar la integridad de estos datos antes de consumirlos en producción.

Paso 1: Inicializar la Sesión de PySpark y Cargar los Datos

Primero, crearemos nuestro punto de entrada a Spark y cargaremos un dataset simulado de transacciones financieras.

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

# Inicializar la sesión de Spark
spark = SparkSession.builder \
    .appName("BigDataDataQualityValidation") \
    .config("spark.sql.execution.arrow.pyspark.enabled", "true") \
    .getOrCreate()

# Definir esquema para optimizar la lectura
schema = StructType([
    StructField("transaction_id", StringType(), True),
    StructField("user_id", StringType(), True),
    StructField("amount", DoubleType(), True),
    StructField("currency", StringType(), True),
    StructField("timestamp", TimestampType(), True)
])

# Cargar datos simulados (reemplaza con la ruta a tu Data Lake)
df = spark.read.schema(schema).parquet("s3a://mi-data-lake-bucket/transacciones/")
print(f"Total de registros cargados: {df.count()}")

Paso 2: Configurar el Contexto de Great Expectations

Great Expectations utiliza un objeto llamado DataContext para organizar las configuraciones, fuentes de datos y expectativas. Vamos a inicializar un contexto en memoria adaptado para trabajar con DataFrames de Spark.

import great_expectations as gx

# Obtener el contexto de datos efímero
context = gx.get_context()

# Agregar el DataFrame de Spark como un Data Source
my_datasource = context.sources.add_spark("mi_datasource_spark")
my_asset = my_datasource.add_dataframe_asset(name="transacciones_asset")

# Crear un Batch Request para procesar el DataFrame
batch_request = my_asset.build_batch_request(dataframe=df)

Paso 3: Definir las Expectativas de Negocio

Las expectativas son afirmaciones comprobables sobre tus datos. Vamos a definir un conjunto de reglas críticas para nuestras transacciones:

1. El identificador de transacción (transaction_id) no debe ser nulo y debe ser único.
2. El monto (amount) debe ser siempre mayor que cero.
3. La moneda (currency) debe pertenecer a un conjunto permitido (USD, EUR, GBP).

# Crear una Suite de Expectativas
suite_name = "transacciones_quality_suite"
suite = context.suites.add(gx.ExpectationSuite(name=suite_name))

# Expectativa 1: transaction_id no debe tener nulos
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(
        column="transaction_id"
    )
)

# Expectativa 2: amount debe estar en un rango lógico (mayor a 0)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount",
        min_value=0.01,
        max_value=1000000.0
    )
)

# Expectativa 3: currency debe ser una de las permitidas
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeInSet(
        column="currency",
        value_set=["USD", "EUR", "GBP"]
    )
)

print(f"Expectativas añadidas exitosamente a la suite: {suite_name}")

📊 Ejecución y Análisis de Resultados

Una vez definidas las reglas, es momento de ejecutar la validación sobre nuestro DataFrame distribuido y procesar el resultado obtenido.

Ejecutando la Validación

# Configurar la definición de validación
validation_definition = context.validation_definitions.add(
    gx.ValidationDefinition(
        data_batch_definition=my_asset.get_batch_definition("batch_def"),
        expectation_suite=suite,
        name="validar_transacciones_diarias"
    )
)

# Ejecutar la validación
validation_result = validation_definition.run(batch_parameters={"dataframe": df})

# Evaluar el éxito general
if validation_result.success:
    print("🎉 ¡Todas las validaciones pasaron exitosamente! Los datos están listos.")
else:
    print("❌ Se detectaron anomalías en los datos. Revise el reporte.")

Interpretando los Resultados

Podemos desglosar el resultado de la validación para entender exactamente qué métricas fallaron y en qué porcentaje.

# Mostrar un resumen de los resultados por expectativa
for result in validation_result.results:
    expectation_type = result.expectation_config.type
    success = result.success
    observed_value = result.result.get("unexpected_percent", 0)
    
    print(f"- Regla: {expectation_type} | ¿Pasó?: {success} | % Inesperado: {observed_value}%")

📋 Buenas Prácticas para Producción

Para llevar este sistema a un entorno de producción robusto, ten en cuenta las siguientes recomendaciones:

❓ Preguntas Frecuentes (FAQs)

🎯 Conclusión

La calidad de los datos ya no es un problema que deba resolverse de forma reactiva tras descubrir un error en producción. Combinando Great Expectations con PySpark, hemos construido una tubería escalable capaz de auditar millones de registros de manera eficiente, asegurando que los productos de datos y modelos de Machine Learning descansen sobre cimientos sólidos y confiables.