Tutorial de Great Expectations: validación de datos con Python

Configuración de Great Expectations

En este tutorial, aprenderás a usar GX Core, la versión de código abierto de Great Expectations, para validar un DataFrame de Pandas. Veremos cómo configurar un contexto, registrar una fuente de datos de Pandas, definir expectativas y validar lotes de datos.

Nota: te recomendamos seguirlo con el notebook de DataLab, aunque también puedes crear tu propio script en Python.

1. Instalar Great Expectations

Requisitos previos

• Python 3.9 a 3.12 instalado.
• Para evitar conflictos, es muy recomendable instalar Great Expectations en un entorno virtual (aviso: la configuración de entornos virtuales queda fuera del alcance de este artículo).
• Un conjunto de datos de ejemplo.

Nota: si usas el notebook de DataLab proporcionado, estos requisitos ya están cubiertos. Puedes saltártelos.

Usa el siguiente comando para instalar GX con pip:

pip install great_expectations

Este comando instala el paquete principal y todas las dependencias necesarias.

2. Inicializar el contexto de datos

Great Expectations necesita un contexto de datos para gestionar configuraciones. Usamos un contexto efímero para evitar persistir configuraciones.

import great_expectations as gx

# Get the Ephemeral Data Context
context = gx.get_context()
assert type(context).__name__ == "EphemeralDataContext"

Crea tu primera suite de validación de datos

Ahora que GX está configurado, vamos a crear una suite de validación.

1. Conectar a una fuente de datos y crear un asset

Una fuente de datos conecta Great Expectations con tus datos, mientras que un data asset representa un subconjunto específico (por ejemplo, una tabla, un DataFrame o un archivo).

En este caso, prepararemos todo para conectarnos a un DataFrame llamado inventory_parts_df. El dataset de ejemplo está disponible en el DataLab proporcionado y se crea al ejecutar el bloque de SQL:

Si no usas DataLab, crea tu propio DataFrame con datos de ejemplo.

Ahora crea la fuente de datos y el asset:

# Add a Pandas Data Source
data_source = context.data_sources.add_pandas(name="inventory_parts")
# Add a Data Asset to the Data Source
data_asset = data_source.add_dataframe_asset(name="inventory_parts_asset")

2. Añadir una definición de lote

Una definición de lote identifica y organiza los datos para su validación. Aquí añadimos una que cubre todo el DataFrame:

# Define the Batch Definition name
batch_definition_name = "inventory_parts_batch"
# Add the Batch Definition
batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name)
assert batch_definition.name == batch_definition_name

3. Recuperar un lote

Un lote es un conjunto de datos asociado a una definición de lote. Para validar, tendrás que recuperar y enlazar el lote con tu DataFrame, en este caso inventory_parts_df:

# Define the Batch Parameters
batch_parameters = {"dataframe": inventory_parts_df}
# Retrieve the Batch
batch = batch_definition.get_batch(batch_parameters=batch_parameters)

4. Crear una suite y definir expectativas

Las expectativas son reglas para validar datos. En este ejemplo, definiremos estas expectativas sencillas:

1. Asegurar que los valores de inventory_id no sean nulos.
2. Asegurar que los valores de part_num sean únicos.

# Create an Expectation Suite
expectation_suite_name = "inventory_parts_suite"
suite = gx.ExpectationSuite(name=expectation_suite_name)
# Add Expectations
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="inventory_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="part_num")
)
# Add the Expectation Suite to the Context
context.suites.add(suite)

Puedes explorar todas las expectativas disponibles en la Expectation Gallery. ¡Te animamos a añadir unas cuantas más!

Después de definirlas, GX devuelve la configuración de la suite de expectativas:

{
  "name": "inventory_parts_suite",
  "id": "b2de0b69-0869-4163-8dde-6c09884483f7",
  "expectations": [
    {
      "type": "expect_column_values_to_not_be_null",
      "kwargs": {
        "column": "inventory_id"
      },
      "meta": {},
      "id": "53d6c42a-d190-412f-a113-783b706531f4"
    },
    {
      "type": "expect_column_values_to_be_unique",
      "kwargs": {
        "column": "part_num"
      },
      "meta": {},
      "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
    }
  ],
  "meta": {
    "great_expectations_version": "1.2.4"
  },
  "notes": null
}

La suite incluye los siguientes detalles:

1. Nombre e ID de la suite: un nombre único (inventory_parts_suite) e identificador para gestionarla y hacer su seguimiento.
2. Expectativas: cada regla especifica:
• El tipo de comprobación (p. ej., asegurar que una columna no tenga valores nulos o que sus valores sean únicos).
• Parámetros, como la columna que se valida.
• Metadatos e ID único por expectativa, lo que facilita su seguimiento y personalización.
3. Metadatos: información de versión de Great Expectations para asegurar compatibilidad.
4. Notas: un espacio para añadir comentarios descriptivos sobre la suite (opcional).

Esta salida estructurada funciona como documentación y como configuración reutilizable para validar tu dataset, de modo que tus expectativas queden claras, trazables y listas para usarlas en el futuro.

5. Validar los datos

Por último, valida el lote con las expectativas definidas y evalúa los resultados.

# Validate the Data Against the Suite
validation_results = batch.validate(suite)
# Evaluate the Results
print(validation_results)

Tras ejecutar la validación, Great Expectations ofrece un informe detallado sobre si el conjunto de datos cumple las expectativas definidas:

{
  "success": false,
  "results": [
    {
      "success": true,
      "expectation_config": {
        "type": "expect_column_values_to_not_be_null",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "inventory_id"
        },
        "meta": {},
        "id": "53d6c42a-d190-412f-a113-783b706531f4"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 0,
        "unexpected_percent": 0.0,
        "partial_unexpected_list": [],
        "partial_unexpected_counts": [],
        "partial_unexpected_index_list": []
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    },
    {
      "success": false,
      "expectation_config": {
        "type": "expect_column_values_to_be_unique",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "part_num"
        },
        "meta": {},
        "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 568352,
        "unexpected_percent": 97.98006788847535,
        "partial_unexpected_list": [
          "48379c01",
          "paddle",
          "11816pr0005",
          "2343",
          "3003",
          "30176",
          "3020",
          "3022",
          "3023",
          "30357",
          "3039",
          "3062b",
          "3068b",
          "3069b",
          "3069b",
          "33291",
          "33291",
          "3795",
          "3941",
          "3960"
        ],
        "missing_count": 0,
        "missing_percent": 0.0,
        "unexpected_percent_total": 97.98006788847535,
        "unexpected_percent_nonmissing": 97.98006788847535,
        "partial_unexpected_counts": [
          {
            "value": "3069b",
            "count": 2
          },
          {
            "value": "33291",
            "count": 2
          },
          {
            "value": "11816pr0005",
            "count": 1
          },
          {
            "value": "2343",
            "count": 1
          },
          {
            "value": "3003",
            "count": 1
          },
          {
            "value": "30176",
            "count": 1
          },
          {
            "value": "3020",
            "count": 1
          },
          {
            "value": "3022",
            "count": 1
          },
          {
            "value": "3023",
            "count": 1
          },
          {
            "value": "30357",
            "count": 1
          },
          {
            "value": "3039",
            "count": 1
          },
          {
            "value": "3062b",
            "count": 1
          },
          {
            "value": "3068b",
            "count": 1
          },
          {
            "value": "3795",
            "count": 1
          },
          {
            "value": "3941",
            "count": 1
          },
          {
            "value": "3960",
            "count": 1
          },
          {
            "value": "48379c01",
            "count": 1
          },
          {
            "value": "paddle",
            "count": 1
          }
        ],
        "partial_unexpected_index_list": [
          0,
          3,
          4,
          5,
          6,
          7,
          8,
          9,
          10,
          11,
          12,
          13,
          14,
          15,
          16,
          17,
          18,
          19,
          20,
          21
        ]
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    }
  ],
  "suite_name": "inventory_parts_suite",
  "suite_parameters": {},
  "statistics": {
    "evaluated_expectations": 2,
    "successful_expectations": 1,
    "unsuccessful_expectations": 1,
    "success_percent": 50.0
  },
  "meta": {
    "great_expectations_version": "1.2.4",
    "batch_spec": {
      "batch_data": "PandasDataFrame"
    },
    "batch_markers": {
      "ge_load_time": "20241129T122532.416424Z",
      "pandas_data_fingerprint": "84a1e1939091fcf54324910def3b89cd"
    },
    "active_batch_definition": {
      "datasource_name": "inventory_parts",
      "data_connector_name": "fluent",
      "data_asset_name": "inventory_parts_asset",
      "batch_identifiers": {
        "dataframe": "<DATAFRAME>"
      }
    }
  },
  "id": null
}

Este informe detalla la calidad de tus datos, señalando aciertos y fallos. Aquí tienes una explicación simplificada de los resultados:

Validación general: el resultado fue parcialmente satisfactorio: el 50% de las expectativas pasaron y el 50% fallaron. Una expectativa fallida indica un problema de calidad que requiere atención. En este caso, una columna no cumple la regla definida.

Expectativa 1: inventory_id no debe tener valores ausentes

• Resultado: aprobado
• Explicación: todos los valores de la columna inventory_id están presentes, sin nulos ni ausencias. Indica buena completitud de datos en esta columna.

Expectativa 2: part_num debe tener valores únicos

• Resultado: no aprobado
• Explicación: la columna part_num contiene un 97,98% de valores duplicados, es decir, muy pocos son únicos.
• Destacados:
• Ejemplos de duplicados: "3069b" y "33291".
• La herramienta también muestra la frecuencia de estos duplicados y sus posiciones de fila, lo que facilita localizarlos y corregirlos.

Por supuesto, este es solo un dataset de ejemplo e incluimos a propósito una expectativa aprobada y otra no aprobada para que veas ambos resultados.

¡Listo! Has ejecutado validaciones de datos de principio a fin.

Integrar Great Expectations en pipelines de datos

En producción, las validaciones deben integrarse directamente en el flujo de trabajo para monitorizar la calidad de los datos de forma continua en cada etapa.

En esta sección, veremos cómo integrar Great Expectations en tus pipelines de datos.

Son ejemplos para orientarte; puede que necesites configuraciones adicionales no incluidas aquí. Consulta la documentación de cada herramienta para la sintaxis actualizada.

Integración con herramientas ETL

Integrar Great Expectations con herramientas ETL populares como Apache Airflow o Prefect es relativamente sencillo. Incluir pasos de validación directamente en los procesos ETL te permitirá detectar y resolver problemas en tiempo real antes de que afecten a los análisis posteriores.

Veamos un ejemplo sencillo de integración con Prefect para ejecutar validaciones como parte de un ETL automatizado:

from prefect import task, Flow
import great_expectations as ge
# Define a task to run Great Expectations validation
@task
def validate_data():
    context = ge.data_context.DataContext()
    batch_kwargs = {"path": "path/to/your/datafile.csv", "datasource": "your_datasource"}
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
	    
    # Check validation results and raise an alert if validation fails
    if not results["success"]:
        raise ValueError("Data validation failed!")
# Define your ETL flow
with Flow("ETL_with_GE_Validation") as flow:
    validation = validate_data()
# Execute the flow
flow.run()

En este ejemplo definimos un flujo de Prefect con una tarea para ejecutar la validación de Great Expectations.

La tarea validate_data() carga el contexto de Great Expectations, recupera el lote de datos y aplica la suite de expectativas.

Si los datos no cumplen los criterios de validación, la tarea lanza una alerta, detiene el flujo y evita errores aguas abajo.

Validación continua de datos

Puedes programar validaciones con diversas herramientas, como cron en sistemas Unix o servicios gestionados como Apache Airflow. En este ejemplo, mostraremos cómo programarlas con Airflow, ideal para orquestar pipelines de datos.

Así puedes configurar un DAG de Airflow (Directed Acyclic Graph) para ejecutar validaciones de Great Expectations a diario:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import great_expectations as ge
# Define the DAG and set the schedule to run daily
default_args = {
	  'owner': 'airflow',
	  'start_date': datetime(2024, 1, 1),
	  'retries': 1,
}
dag = DAG(
      'great_expectations_validation',
	default_args=default_args,
	schedule_interval='@daily',  # Runs once a day
)
# Define the function to run the validation
def run_validation():
    context = ge.data_context.DataContext()
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
    return results
# Set up the task in Airflow
validation_task = PythonOperator(
      task_id='run_great_expectations_validation',
      python_callable=run_validation,
      dag=dag,
)
# Set the task in the DAG
validation_task

En este ejemplo definimos un DAG que programa una validación una vez al día (@daily).

La función run_validation() ejecuta la validación cargando el contexto de Great Expectations y aplicando la suite de expectativas definida a los datos.

Buenas prácticas para validar datos con Great Expectations

Seguir buenas prácticas siempre es aconsejable para escalar y ser eficiente, y con Great Expectations no es diferente.

Empieza poco a poco e itera

Comienza con comprobaciones básicas de calidad y amplía gradualmente. Es mejor centrarse primero en expectativas fundamentales para no complicar en exceso el proceso, lo que facilita la integración y el diagnóstico. A medida que entiendas mejor el dataset, podrás añadir validaciones más complejas.

Colabora entre equipos

La calidad de datos no es solo un asunto técnico. Colabora con equipos de negocio para definir expectativas y asegurarte de que la validación implementada esté alineada con la lógica y los objetivos del negocio. Este enfoque transversal garantiza que los datos cumplan su propósito y las necesidades de todos los stakeholders.

Automatiza siempre que puedas

Automatiza el proceso siempre que sea posible para integrar la validación en los pipelines. Incluir comprobaciones automáticas permite monitorizar la calidad de forma continua sin intervención manual, lo que mejora mucho la eficiencia.

Conclusión

¡Buen trabajo! Has aprendido a configurar y validar datos en Great Expectations. Estas técnicas te ayudarán a mantener una alta calidad y transparencia en tus flujos de trabajo.

Para seguir desarrollando tus habilidades, consulta estos recursos:

• ETL and ELT in Python: aprende a transformar y mover datos de forma eficiente.
• Introduction to Data Quality: descubre los fundamentos de la gestión de la calidad de datos.
• Cleaning Data in Python: domina técnicas de limpieza para asegurar precisión y coherencia.
• Data Quality Dimensions Cheat Sheet: una guía práctica de las dimensiones de calidad de datos.