En ingeniería de datos, si entra basura, sale basura (Garbage In, Garbage Out). Muchas veces confiamos en que los CSV que recibimos vienen limpios, pero la realidad es que suelen traer fechas mal formadas, montos negativos o IDs faltantes que rompen nuestros pipelines en BigQuery o bases de datos de producción.
En este post documento una solución ligera y potente: usar DuckDB como motor de validación local mediante SQL, gestionando las reglas en un archivo JSON externo para que el sistema sea escalable.
El problema de la validación tradicional #
Validar datos puramente en Python con Pandas puede volverse un caos de if/else y lógica difícil de mantener. Al usar DuckDB, aprovechamos la potencia declarativa de SQL para detectar anomalías en milisegundos, incluso con archivos de gigabytes.
Datos de prueba (data.csv) #
Crea un archivo de ejemplo para probar el script. Nota cómo incluimos errores intencionales (edad negativa, ID vacío):
Datos de ejemplo #
ID,Nombre,Edad,Email,FechaPedido,MontoVenta
1,Alice,25,[email protected],2023-01-01,100.50
2,Bob,30,[email protected],2023-03-15,200.00
3,Charlie,-5,[email protected],2023-05-01,50.75
,David,40,[email protected],2023-07-20,0.00
5,Eve,30,[email protected],2025-01-01,150.25
6,Frank,22,[email protected],2023-09-10,-20.00
7,Grace,28,[email protected],2023-11-05,300.00
Reglas de validación #
Lo ideal es separar la lógica del código. Aquí definimos qué queremos encontrar (los errores):
[
{
"name": "ID faltante",
"query_sql": "SELECT * FROM data_to_validate WHERE ID IS NULL"
},
{
"name": "Edad incorrecta",
"query_sql": "SELECT * FROM data_to_validate WHERE Edad < 0"
},
{
"name": "Monto negativo",
"query_sql": "SELECT * FROM data_to_validate WHERE MontoVenta < 0"
}
]
Script #
Este script carga el CSV, registra las reglas y nos entrega un reporte de integridad.
import pandas as pd
import duckdb
import json
def validate_csv_with_duckdb(csv_path: str, validations_path: str):
print("\n--- INICIANDO VALIDACIÓN DE DATOS ---")
try:
df = pd.read_csv(csv_path)
print(f"CSV '{csv_path}' cargado exitosamente.")
except FileNotFoundError:
print(f"ERROR: Archivo CSV no encontrado en '{csv_path}'")
return
except Exception as e:
print(f"ERROR al cargar CSV: {e}")
return
try:
with open(validations_path, 'r', encoding='utf-8') as f:
validations = json.load(f)
print(f"Cargadas {len(validations)} validaciones desde '{validations_path}'.")
except FileNotFoundError:
print(f"ERROR: Archivo de validaciones no encontrado en '{validations_path}'")
return
except json.JSONDecodeError:
print(f"ERROR: El archivo '{validations_path}' no es un JSON válido.")
return
# Conectar DuckDB y registrar el DataFrame
con = duckdb.connect(database=':memory:', read_only=False)
con.register('data_to_validate', df) # Registramos el DF con el nombre que usamos en las validacione JSON
failed_validations = []
# Ejecutar cada validación y recolectar fallos
for validation in validations:
name = validation.get('name', 'Validación sin nombre')
query_sql = validation.get('query_sql')
if not query_sql:
print(f"ADVERTENCIA: Validación '{name}' sin query SQL. Saltando.")
continue
try:
# Ejecutar la query. Si devuelve filas, la validación falló.
failed_records = con.execute(query_sql).fetchdf()
if not failed_records.empty:
failed_validations.append(name)
print(f" FALLÓ: '{name}' - Se encontraron {len(failed_records)} problemas.")
except Exception as e:
print(f"ERROR al ejecutar query para '{name}': {e}")
failed_validations.append(f"{name} (ERROR SQL)")
# No queremos que un error en una query detenga todo, solo lo reportamos.
con.close() # Cerrar la conexión a DuckDB
# Resultado final
print("\n--- REPORTE FINAL ---")
if not failed_validations:
print("\033[92m✅ ¡Todas las validaciones pasaron correctamente! Los datos están limpios.\033[0m") # Verde
else:
print("\033[91m❌ Se encontraron problemas de validación en los siguientes puntos:\033[0m") # Rojo
for validation_name in failed_validations:
print(f" - {validation_name}")
print("\n\033[93mPor favor, revisa los datos y las validaciones fallidas.\033[0m") # Amarillo
print("\n--- VALIDACIÓN FINALIZADA ---")
if __name__ == "__main__":
CSV_FILE = '/content/sample_data/data.csv'
VALIDATIONS_FILE = '/content/sample_data/validations.json'
validate_csv_with_duckdb(CSV_FILE, VALIDATIONS_FILE)
Domina la lógica detrás de DuckDB
DuckDB es increíblemente rápido, pero su verdadero poder depende de qué tan bueno sea tu SQL. Si todavía te trabas con las queries complejas o el manejo de datos anidados, te sugiero echarle un ojo a los tracks de Data Engineering en DataCamp.
Nota de transparencia: El siguiente es un enlace de afiliado. Si decides suscribirte a través de él, yo recibo una comisión que ayuda a mantener este blog libre de publicidad invasiva, sin que a ti te cueste un centavo extra.
Explorar DataCamp →Nota: DuckDB es el “SQLite de la analítica”. La ventaja de este enfoque es que puedes correr estas validaciones en una Raspberry Pi o en un contenedor ligero de Docker sin necesidad de levantar una base de datos pesada.
No te quedes solo con la teoría
En el canal de YouTube subo los videos de lo que voy aprendiendo y construyendo.
Ir al canal →¿Te quedó alguna duda o quieres sugerir un tema? Te leo en los comentarios del video.
¿Te ha sido útil? Puedes invitarme un café para apoyar la creación de más contenido gratuito.
