Cómo validar la calidad de los datos en un pipeline ETL

Un pipeline ETL que carga datos erróneos es peor que no tener pipeline: los errores se propagan en silencio hasta los informes y las decisiones. Emails vacíos, edades imposibles o clientes repetidos acaban en la tabla final sin que nadie se dé cuenta. La validación de la calidad de los datos resuelve el problema comprobando cada fila antes de la carga y dejando pasar solo las que cumplen tus reglas. Es un paso sencillo de añadir con Python y pandas, y marca una gran diferencia en la confianza que tienes en tus datos.

Requisitos previos

• Python 3.9 o superior instalado en tu ordenador.
• La biblioteca pandas instalada (pip install pandas).
• Un archivo CSV de ejemplo para validar (usamos clientes.csv).
• Nociones básicas del flujo ETL: extraer, transformar y cargar.

Paso 1: Extraer los datos y definir las reglas

El primer paso es leer los datos de origen en un DataFrame y escribir, de forma explícita, las reglas que cada fila debe cumplir. Tener las reglas reunidas en un único lugar hace que el pipeline sea fácil de mantener y de auditar: si mañana cambia la regla de la edad, editas un único número. En nuestro ejemplo tenemos una tabla de clientes y cuatro reglas de calidad: el email no puede estar vacío, la edad debe estar entre 18 y 120, el país debe pertenecer a una lista permitida y el id debe ser único.

import pandas as pd

# Extrair (E): ler os dados de origem
df = pd.read_csv("clientes.csv")

# Lista de valores permitidos para o pais
PAISES_VALIDOS = ["PT", "ES", "FR", "DE"]

Paso 2: Escribir la función de validación

En lugar de borrar de inmediato las filas inválidas, vamos a anotar el motivo de cada fallo. Así guardamos un registro claro de lo que salió mal, lo que ayuda a corregir el origen más tarde. La función siguiente recibe una fila y devuelve una lista con todos los problemas encontrados — fíjate en que reunimos todos los errores, y no solo el primero. De esta forma, quien corrige los datos ve todos los problemas de una vez en lugar de descubrirlos uno a uno.

def validar_linha(linha):
    problemas = []
    if pd.isna(linha["email"]) or str(linha["email"]).strip() == "":
        problemas.append("email vazio")
    if pd.isna(linha["idade"]) or linha["idade"] < 18 or linha["idade"] > 120:
        problemas.append("idade invalida")
    if linha["pais"] not in PAISES_VALIDOS:
        problemas.append("pais nao permitido")
    return problemas

Paso 3: Aplicar las reglas y separar las filas

Ahora aplicamos la función a todas las filas con apply y axis=1, y añadimos la comprobación de ids duplicados. A diferencia de las otras reglas, detectar duplicados implica comparar cada id con todos los demás, por lo que se ejecuta sobre la columna completa y no fila a fila. El resultado es una columna erros con el texto de los problemas de cada fila. Las filas sin errores son las válidas; el resto va a cuarentena.

# Erros por linha (regras individuais)
df["erros"] = df.apply(validar_linha, axis=1)

# Regra que envolve a coluna toda: id duplicado
duplicados = df["id"].duplicated(keep=False)
df.loc[duplicados, "erros"] = df.loc[duplicados, "erros"].apply(
    lambda lista: lista + ["id duplicado"]
)

# Transformar a lista de erros em texto
df["erros"] = df["erros"].apply(lambda lista: ", ".join(lista))

# Separar validas de invalidas
validas = df[df["erros"] == ""].copy()
invalidas = df[df["erros"] != ""].copy()

Paso 4: Cargar las válidas y guardar las inválidas

En la fase de carga (L) enviamos solo las filas válidas al destino. Las inválidas no se descartan: se guardan en un archivo de cuarentena con la columna erros, para que alguien pueda revisarlas y corregirlas. Aquí usamos archivos CSV para mantener el ejemplo simple, pero el destino podría ser una base de datos o un data warehouse.

# Carregar (L): so as linhas validas
validas.drop(columns=["erros"]).to_csv("clientes_limpos.csv", index=False)

# Quarentena: linhas invalidas com o motivo
invalidas.to_csv("clientes_quarentena.csv", index=False)

print(f"Validas: {len(validas)} | Invalidas: {len(invalidas)}")

Verificar el resultado

Para confirmar que la validación funcionó, abre el archivo clientes_quarentena.csv y comprueba que cada fila rechazada tiene una explicación en la columna erros. El número de filas válidas más el de inválidas debe ser igual al total leído al inicio — la línea assert lo confirma automáticamente y falla si las cuentas no cuadran.

# Verificacao rapida: as contas fecham?
assert len(validas) + len(invalidas) == len(df)
print(invalidas[["id", "erros"]])

Conclusión

Con solo unas pocas líneas de código ya tienes un pipeline ETL que confía, pero verifica: las filas buenas siguen hacia el destino y las dudosas quedan en cuarentena con el motivo. A partir de aquí puedes añadir nuevas reglas (formato del email, fechas futuras, valores negativos), convertir esta comprobación en un paso reutilizable o registrar métricas de calidad en cada ejecución. ¿Qué regla de calidad es la más importante para tus datos, y ya la estás validando?