Introducción

Recientemente, trabajé en un proyecto de Ingeniería de Datos para construir una arquitectura de Data Warehouse, ingestando datos CSV de múltiples fuentes. Tuvimos la oportunidad de implementar controles de Data Quality en múltiples pipelines de ETL. Existen varias herramientas de calidad de datos basadas en Python para lograr este objetivo.

En este artículo, quiero revisar conceptos fundamentales de Data Quality y compartir dos de estas herramientas, concretamente Great Expectations y PyDeequ, que consideramos para implementar los controles. Mostraré algunas ventajas y desventajas que observamos en cada una, para ayudarte si es la primera vez que te disponés a implementar esta funcionalidad.

¿Por qué Data Quality?

Descuidar la calidad de los datos puede tener una serie de consecuencias negativas que afectan el rendimiento, la reputación y la competitividad de tu organización. Si no hay datos de buena calidad, los análisis y decisiones pueden resultar inexactos. Los stakeholders podrían tomar decisiones basadas en información desactualizada o errónea, lo que podría llevar a decisiones equivocadas y errores potencialmente costosos. Es inevitable que, al identificar la fuente de errores en los datos, la confianza en los datos y en los equipos responsables se vea erosionada. Además, si los datos se comparten fuera de la organización y hay problemas de Data Quality, los usuarios también pueden perder la confianza en ellos, dañando la reputación de la organización.

Además, trabajar con datos de baja calidad puede llevar a percepciones incorrectas y a una pérdida de enfoque en el verdadero valor de la información, haciendo que los equipos de datos pierdan tiempo buscando la fuente de los errores. Mirándolo de manera positiva: asegurar la calidad de los datos significa que las decisiones en tu organización estarán bien informadas y actualizadas, evitando costos asociados con decisiones incorrectas y mejorando la competitividad de tu empresa.

¿Cuáles son los aspectos de Data Quality que solemos evaluar?

• Exactitud (Accuracy): La exactitud de los datos mide cuán bien reflejan los datos las entidades del mundo real que representan. Los datos precisos están libres de errores e inconsistencias.
• Integridad (Completeness): La integridad evalúa si todos los datos necesarios están presentes. Asegura que no falte información crítica en los conjuntos de datos.
• Oportunidad-puntualidad (Timeliness): Timeliness se refiere a la frescura de los datos. Mide cuán actualizados están los datos y si cumplen con los plazos requeridos para análisis y toma de decisiones.
• Unicidad (Uniqueness): La unicidad garantiza que no haya registros duplicados en el conjunto de datos.
• Validez (Validity): La validez verifica si los datos cumplen con reglas y restricciones predefinidas; asegura que los datos estén en el formato esperado y cumplan con estándares de calidad.
• Coherencia (Consistency): La coherencia evalúa la uniformidad de los datos en diferentes fuentes y sistemas. Datos inconsistentes pueden llevar a confusiones e inexactitudes.
• Aptitud para el propósito (Fitness for Purpose):  Los datos deben ser aptos para su propósito previsto. Esto significa que los datos deben ser relevantes y adecuados para tareas y análisis específicos.

Principales características de las Herramientas de Data Quality 

Algunas de las características clave de las herramientas de Data Quality que podés usar para elegir la que te sea más conveniente son:

• Validación de Datos: Un elemento esencial de las herramientas de Calidad de Datos es la capacidad de verificar que se cumplan ciertas “reglas” en los datos que queremos analizar. Esto incluye la capacidad de verificar si los valores de las columnas están dentro de rangos específicos, si no hay valores nulos en una columna, si los valores estadísticos como promedios y desviaciones estándar están dentro de límites predefinidos, entre otras validaciones.
• Profiling / Sugerencia Automática de Reglas de Validación: Otra característica importante de las herramientas de Data Quality es la capacidad de realizar un análisis automático de una porción de los datos que queremos procesar. Esto permite extraer características generales de los datos, identificar los rangos en los que se encuentran las variables, determinar los tipos de variables, detectar valores faltantes e identificar patrones de distribución. Con base en esta información (perfilado), las herramientas pueden sugerir posibles validaciones de Calidad de Datos y reglas que se pueden aplicar.
• Detección de Anomalías: Basándose en el perfil o comportamiento de las columnas en nuestro conjunto de datos, estas herramientas pueden proponer alertas que detecten si alguna métrica relacionada con los datos, como el promedio o la desviación estándar, están fuera de límites predefinidos y se desvían significativamente de lo esperado. Esta función es esencial para identificar datos atípicos o inusuales que podrían indicar problemas de calidad de los datos o problemas en el proceso de adquisición de los mismos.
• Generación de Informes: Algunas herramientas de Calidad de Datos pueden generar visualizaciones interactivas, informes detallados o resúmenes ejecutivos que sintetizan los resultados del análisis del conjunto de datos y las validaciones aplicadas. Estos informes ayudan a los usuarios a comprender la Calidad de Datos y los hallazgos clave de manera clara y concisa, facilitando la toma de decisiones informadas y la comunicación de resultados a las partes interesadas.
• Recuperación de Registros Fallidos: En caso de que no se cumplan las validaciones, algunas herramientas de Data Quality permiten a los desarrolladores identificar registros específicos que contienen errores, los cuales pueden marcarse y priorizarse para su revisión y acción. Esto ayuda a mantener la integridad de los datos y garantiza que los problemas se aborden de manera oportuna.
• Integraciones: Muchas herramientas de Data Quality ofrecen diversas opciones para conectarse a otras fuentes de datos, sistemas de orquestación de procesos o para enviar resultados a varios destinos. Estas integraciones facilitan la automatización del flujo de trabajo, la incorporación de datos externos y la distribución eficiente de resultados de análisis a través de diferentes canales y sistemas, mejorando la eficacia de las operaciones relacionadas con la Calidad de Datos.

PyDeequ

Descripción

PyDeequ es una biblioteca de validación de datos originalmente diseñada en Scala y posteriormente adaptada para su uso en Python. Esta herramienta proporciona capacidades para la detección de anomalías, profiling y la sugerencia de controles de Data Quality. PyDeequ se integra con el ecosistema Spark, lo que permite la gestión eficiente de datasets de gran tamaño.

En PyDeequ, la funcionalidad de sugerencia de reglas (constraint suggestion) te permite analizar una porción de los datos para obtener rápidamente condiciones de Calidad de Datos que se aplicarán a cada columna. Estas restricciones pueden ser posteriormente copiadas e incorporadas en un conjunto de verificaciones encadenadas para utilizarlas en el resto de los datos a procesar. Los constraints también te permiten especificar un umbral porcentual en el cual se debe activar un error o advertencia mediante funciones lambda. También podés especificar pistas o hints para identificar rápidamente los constraints cuando se entregan los resultados.

Además, PyDeequ proporciona varios analizadores que sirven como medidas de características específicas del conjunto de datos, como el tamaño del dataset, desviación estándar, porcentaje de valores únicos, matching de patrones definidos por expresiones regulares, etc. Estos analizadores se pueden utilizar para detectar anomalías basadas en un conjunto de datos de referencia estableciendo umbrales. Si los valores del analizador varían entre conjuntos de datos más allá de una cierta proporción definida por los umbrales, se informa este fenómeno.

Guía de Inicio Rápido

1. Generá un conjunto de datos de ejemplo (podés ejecutar este código en el directorio de trabajo y los datos se generarán en Example_Data/Batches/Example_batch0.csv):

from faker import Faker
import pandas as pd
import random

fake = Faker(seed=123) 

num_rows = 1000

data = {
    "Name": [fake.name() for _ in range(num_rows)],
    "Email": [fake.email() for _ in range(num_rows)],
    "Phone": [fake.phone_number() for _ in range(num_rows)],
    "Address": [fake.address() for _ in range(num_rows)],
    "Date of Birth": [fake.date_of_birth(minimum_age=18, maximum_age=80).strftime("%Y-%m-%d") for _ in range(num_rows)],
    "Salary": [fake.random_int(min=30000, max=100000) for _ in range(num_rows)],
    "Department": [fake.random_element(elements=("HR", "IT", "Finance", "Sales", "Marketing")) for _ in range(num_rows)],
    "Hire Date": [fake.date_this_century().strftime("%Y-%m-%d") for _ in range(num_rows)],
    "Employee ID": [fake.unique.random_number() for _ in range(num_rows)],
"Vacation_Days": [fake.random_int(min=0, max=30) if random.random() > 0.1 else None for _ in range(num_rows)]
}

# Introduce null values in some columns
for i in range(100):
    data["Email"][random.randint(0, num_rows - 1)] = None
    data["Salary"][random.randint(0, num_rows - 1)] = None

# Replace some values in the "Department" column with rare values
for i in range(50):
    data["Department"][random.randint(0, num_rows - 1)] = fake.random_element(
        elements=("Legal", "R&D", "Customer Support")
    )
    
df = pd.DataFrame(data) # Create a DataFrame from the generated data

output_folder = "Example_Data/Batches/"
os.makedirs(output_folder, exist_ok=True)

split_dfs = np.array_split(df, 3)  # Split the DataFrame

for i, split_df in enumerate(split_dfs):
    file_name = f"{output_folder}Example_batch{i}.csv"
    split_df.to_csv(file_name, sep=",", index=False)

2. Cargá el conjunto de datos de ejemplo y creá una sesión de Spark.

import pandas as pd
from pyspark.sql import SparkSession
import sagemaker_pyspark
import pydeequ  

df = pd.read_csv('Example_Data/Batches/Example_batch0.csv')

classpath = ":".join(sagemaker_pyspark.classpath_jars())

spark = (SparkSession.builder
	.config("spark.driver.extraClassPath", classpath)
	.config("spark.jars.packages", pydeequ.deequ_maven_coord)
	.config("spark.jars.excludes", pydeequ.f2j_maven_coord)
	.getOrCreate()
	)

3. Corré el comando que sugiere los constraints.

from pydeequ.suggestions import *  

suggestionResult = ConstraintSuggestionRunner(spark) \
	.onData(df) \
	.addConstraintRule(DEFAULT()) \
	.run()

4. Podés mostrar los constraints sugeridos:

for sugg in suggestionResult['constraint_suggestions']:
	print(f"Constraint suggestion for \'{sugg['column_name']}\': {sugg['description']}")

for sugg in suggestionResult['constraint_suggestions']:
	print(f"{sugg['code_for_constraint']}")

5. Cargá un nuevo batch de datos:

df = pd.read_csv('Example_Data/Batches/Example_batch1.csv')
df.columns

6. Creá un check:

from pydeequ.checks import *
from pydeequ.verification import *
	
check = Check(spark, CheckLevel.Error, "Integrity checks")

7. Añadí los constraints sugeridos al check, formando una “verification suite,” y correlo:

checkResult = VerificationSuite(spark) \
.onData(df) \
.addCheck(
check.hasSize(lambda x: x <= 300_000) \
	.isContainedIn("Department", ["Marketing", "HR", "Finance", "IT", "Sales"])
	.isComplete("Department")
	.isComplete("Name")
	.isUnique("Name")
	.isComplete("Address")
	.isUnique("Address")
	.isComplete("Hire_Date")
	.isUnique("Hire_Date")
	.isComplete("Vacation_Days")
	.isNonNegative("Vacation_Days")
	.isComplete("Date_of_Birth")
	.isComplete("Email")
	.isUnique("Email")
	.isComplete("Phone")
	.isComplete("Salary")
	.isNonNegative("Salary")
	.isComplete("Employee_ID")
	.isNonNegative("Employee_ID")
	.isUnique("Employee_ID")
).run()

8. Convertí los resultados a un dataframe de Pandas y mostralos:

checkResult_df = VerificationResult.checkResultsAsDataFrame(spark, checkResult)
checkResult_df.toPandas().head(10)

Hay que tener cuidado de no usar espacios entre los nombres de columnas o tener errores de tipeo en ellos, ya que esto puede llevar a mensajes de error en muchas de las condiciones. Voy a mostrar este error corriendo la “verification suite” de nuevo, pero borrando el guión bajo de “Employee_ID” en la línea que tiene el check .isNonNegative(“Employee_ID”). El resultado es:

Comentarios adicionales:

• Una característica única de PyDeequ es el constraint “satisfies”, que permite definir condiciones en lenguaje SQL para las columnas. Esto es especialmente útil para validar que los datos cumplan condiciones lógicas compuestas por múltiples condiciones más simples.
• Es importante señalar que, a diferencia de algunas otras herramientas, PyDeequ no genera informes; entrega resultados en un dataframe, y el usuario es responsable de implementar la funcionalidad para acceder a los datos e incorporarlos en un informe.

Great Expectations

Descripción:

Great Expectations es una herramienta de código abierto basada en Python que cuenta con el respaldo de una gran comunidad. Este proyecto se ha integrado con éxito con una amplia variedad de fuentes y herramientas de datos, convirtiéndolo en una elección versátil para la gestión de Data Quality.

En Great Expectations, las reglas de Data Quality se estructuran como “expectations”. Estas expectations se definen “de manera positiva”, lo que significa que especifican las condiciones que los datos deben cumplir. Si los datos no cumplen con estas condiciones, la expectation devuelve una alerta, señalando un posible problema de Data Quality.

Este enfoque permite que los profesionales de datos articulen y hagan cumplir los requisitos de Calidad de Datos de manera clara y práctica. Por ejemplo, se pueden establecer expectations de que las columnas de datos deben cumplir con tipos de datos específicos, rangos o restricciones de valores. Si los datos se desvían de estas expectativas, Great Expectations generará las alertas correspondientes.

Además, Great Expectations ofrece otras funciones que incluyen el profiling de datos, validación automática, monitoreo continuo e integración con diversas plataformas de datos. Se distingue sobre todo al proporcionar una buena documentación de los datos, ofreciendo información valiosa sobre distribuciones de datos y estadísticas resumidas. Esta documentación no solo mejora la transparencia de los datos, sino que también facilita la colaboración entre equipos de datos.

Si bien Great Expectations ofrece muchas ventajas, los usuarios tienen que tener muy presente su documentación, ya que las actualizaciones frecuentes pueden generar cambios y variaciones entre versiones. Además, expresar condiciones lógicas complejas directamente mediante expectations puede volverse algo complejo. Sin embargo, la herramienta proporciona la flexibilidad de trabajar con “row conditions” para manejar lógica más intrincada.

A continuación vamos a ver algunos comandos para entender las funcionalidades clave de esta herramienta.

Guía de Inicio Rápido

Para comenzar, en un entorno Conda con Python, podés instalar Great Expectations usando pip con el siguiente comando:

pip install great_expectations
pip install great_expectations_experimental

Personalmente, recomiendo empezar utilizando la interfaz de línea de comandos (CLI) de Great Expectations para realizar las configuraciones iniciales de nuestro proyecto. Esta herramienta nos guía paso a paso en la definición de las fuentes de datos y las validaciones que queremos realizar, y crea automáticamente cuadernos (notebooks) con el código necesario para implementarlas. Con ese código, más adelante podemos realizar cambios para determinar completamente las operaciones y obtener los resultados de validación en el formato esperado.

Para desarrollar un proyecto utilizando la CLI, hay que crear un directorio vacío y activar el entorno Conda donde se instaló Great Expectations. A partir de ahí, hay que usar los siguientes comandos.

great_expectations init

great_expectations init crea una estructura de directorios para Great Expectations, donde se almacenarán las fuentes de datos (datasources) que señalan hacia los datos que se quieren procesar, las reglas de validación que se aplicarán, las notebooks generadas por el asistente y los informes HTML resultantes de los análisis.

A continuación, al ejecutar great_expectations datasource new, se especifica desde dónde se recibirá el dataset a procesar. Entre las opciones, se puede elegir Sistema de Archivos (File System) y un servidor SQL. Si se selecciona Sistema de Archivos, hay que determinar si el motor utilizado es Spark o Pandas y especificar el directorio donde se encuentran los archivos a procesar (todo esto se muestra en la imagen a continuación).

Una vez confirmados los pasos anteriores, el asistente abrirá una Jupyter notebook que debe ejecutarse por completo para que las configuraciones se apliquen al proyecto. La notebook se encuentra en el directorio del proyecto y se puede optar por eliminarla o guardarla para realizar cambios de configuración más adelante.

En la Jupyter notebook, hay que revisar el string yaml de la configuración del Datasource. En el campo “data_connectors”, las primeras líneas deberían tener información como:

El siguiente comando es great_expectations suite new, con el cuál se crea una nueva suite de expectations. El asistente ofrece 3 opciones:

• Crear la suite manualmente
• Crear la suite a partir de un data asset
• Crear la suite con el asistente.

Si es la primera vez que usás Great Expectations, te recomiendo la opción 3, ya que las 2 primeras requieren conocimiento de las expectations disponibles, y conocimiento de dominio para determinar qué validaciones se necesitan para nuestros datos. Si seleccionaste la opción 3, podés elegir un archivo dentro del directorio de datos que se configuró previamente, para inferir el esquema (schema) de las tablas y las validaciones que podrían aplicarse.

Cuando seleccionás el archivo elegido para inferir el esquema, se te va a pedir que asignes un nombre a la nueva suite y confirmes la selección final. Después de esto, se abre una notebook que hay que ejecutar para que Great Expectations sugiera las expectations que se pueden aplicar al dataset que especificaste. Hay que ejecutar todas las celdas en el cuaderno, y se va a abrir entonces una nueva pestaña en el navegador con un informe HTML que describe las expectations generadas para la suite.

Estas expectations se pueden encontrar en el directorio del proyecto en “great_expectations/expectations/<nombre_de_la_suite>.json”. Se puede modificar este archivo añadiendo o eliminando expectations. El catálogo de expectations disponibles, junto con su estado de desarrollo, se puede encontrar en https://greatexpectations.io/expectations/.

Ejecutar great_expectations checkpoint new <nombre_del_checkpoint> abre una notebook que permite configurar el checkpoint. En esta notebook, se enumeran muchas configuraciones basadas en el datasource y la suite de expectations configuradas en los pasos anteriores. También permite establecer configuraciones adicionales. Por ejemplo, se puede elegir un archivo diferente al original al cual se deben aplicar las expectations. Además, con un código adicional, se puede especificar que las expectations se apliquen a varios archivos cuyos nombres coincidan con una cierta expresión regular (multi-batch request).

La última celda en la notebook contiene líneas de código comentadas que, al ejecutarse, aplican el checkpoint y muestran los resultados en un informe HTML en una nueva pestaña del navegador. Si utilizamos un batch de datos sin una de las columnas y con filas faltantes, deberíamos ver un informe como el que se muestra a continuación:

Los checkpoints que se crearon quedan en el directorio del proyecto en: “great_expectations/checkpoints/<checkpoint_name>.yml”. Los resultados de los checkpoints se pueden encontrar en el directorio del proyecto en formato JSON en:

great_expectations/uncommitted/validations/<suite_name>/<time_stamp>/<time_stamp>/<random_hash_name>.json

o bien como reportes en formato HTML en

great_expectations/uncommitted/data_docs/local_site/validations/<suite_name>/<time_stamp>/<time_stamp>/<random_hash_name>.HTML

Finalmente, con great_expectations checkpoint run <nombre_del_checkpoint> se puede aplicar el checkpoint a través de la CLI

Comentarios adicionales:

• Te recomiendo mirar el workflow de multi-batch requests.
• Entre las integraciones disponibles en Great Expectations, quiero subrayar las siguientes:
  • Conexión a bases de datos SQL
  • Conexión a almacenamiento cloud (S3, CGS, Azure Blob Storage)
  • Orquestación con Airflow

Conclusión

En este artículo, pudimos ver los fundamentos de la Calidad de Datos e introducir dos herramientas basadas en Python para realizar análisis en este campo en nuestros datos. PyDeequ adopta un enfoque algo más simple y directo para aplicar controles, mientras que Great Expectations ofrece características muy útiles para proyectos de escala algo más grande. Si estás interesado en aprender más sobre estas herramientas, dejá tus preguntas en los comentarios.