Tu primer pipeline de datos en Python: de un CSV sucio a una tabla limpia

El pipeline más simple del mundo: leer, limpiar, guardar. Todo lo demás es decoración sobre esta idea.

Tienes un CSV que te ha pasado alguien de negocio. Lo abres y ves fechas en tres formatos distintos, filas duplicadas, celdas vacías donde debería haber números y una columna que se llama "Unnamed: 0". Suena familiar, ¿verdad? Bienvenido al 80% del trabajo real con datos.

Vamos a construir un mini pipeline — un programa que lea ese CSV, lo limpie y lo deje listo para analizar. Sin frameworks, sin Docker, sin Airflow. Solo Python y pandas. Cuando entiendas esto, entenderás lo que hace cualquier herramienta de ETL por debajo: es la misma idea con más botones.

### Qué es un pipeline de datos (en 30 segundos)

Un pipeline es un programa que mueve datos de un sitio A a un sitio B, transformándolos por el camino. La analogía clásica: una planta potabilizadora. El agua entra sucia del río (Extract), se filtra y se trata (Transform), y sale limpia por el grifo (Load). ETL. Tres letras, una idea.

Nuestro pipeline va a hacer exactamente eso con un fichero de ventas: leerlo tal cual llega, arreglar lo que esté roto y guardarlo en un formato que cualquier herramienta analítica pueda consumir sin problemas.

### El CSV sucio: nuestro punto de partida

Vamos a simular un fichero realista. Tiene los problemas habituales: duplicados, nulos, tipos incorrectos y alguna sorpresa. Así es como se ve el mundo real — nadie te pasa un dataset limpio.

1# ventas_2024.csv (simulado)
2# fecha,producto,cantidad,precio_unitario,tienda
3# 2024-01-15,Laptop,2,899.99,Madrid
4# 15/01/2024,Laptop,2,899.99,Madrid      ← duplicado con formato distinto
5# 2024-01-16,Monitor,,349.50,Barcelona   ← cantidad vacía
6# 2024-01-17,Teclado,5,29.99,
7# 2024-01-17,Ratón,abc,19.99,Valencia    ← "abc" no es un número
8# ...
9
10import pandas as pd
11
12# --- EXTRACT ---
13df = pd.read_csv('ventas_2024.csv')
14print(f"Filas crudas: {len(df)}")
15print(df.dtypes)

Paso 1: leer el CSV tal cual. Todavía no arreglamos nada — primero miramos.

Siempre mira los datos ANTES de tocarlos. Un df.head(), df.info() y df.describe() te ahorran horas de debuggeo. El 90% de los bugs en un pipeline vienen de asumir cómo son los datos sin haberlos mirado.

### Paso 1: eliminar duplicados

Los duplicados son el problema más común y el más fácil de resolver. A veces son filas idénticas (alguien ejecutó la carga dos veces), a veces son la misma venta con el formato de fecha cambiado. Para el primer caso, pandas tiene una función directa.

1# --- TRANSFORM: paso 1 — duplicados ---
2antes = len(df)
3df = df.drop_duplicates()
4despues = len(df)
5print(f"Duplicados eliminados: {antes - despues}")
6
7# Si la duplicación es por formato de fecha distinto,
8# primero normalizas la fecha y DESPUÉS eliminas:
9# df['fecha'] = pd.to_datetime(df['fecha'], dayfirst=False, errors='coerce')
10# df = df.drop_duplicates(subset=['fecha', 'producto', 'tienda'])

drop_duplicates() compara fila completa por defecto. Con subset, eliges las columnas clave.

### Paso 2: gestionar nulos

Los nulos son el "depende" de los datos. A veces los eliminas, a veces los rellenas, y a veces los dejas. La decisión depende del contexto: un nulo en "cantidad" probablemente es un error de carga (lo descartas o preguntas), un nulo en "tienda" puede significar venta online.

1# --- TRANSFORM: paso 2 — nulos ---
2print(df.isnull().sum())  # ver cuántos hay por columna
3
4# Estrategia: 
5# - cantidad nula → descartar (no podemos inventar ventas)
6# - tienda nula → rellenar con "online"
7df = df.dropna(subset=['cantidad'])
8df['tienda'] = df['tienda'].fillna('online')
9
10print(f"Filas tras limpiar nulos: {len(df)}")

No existe una regla universal para nulos. La regla es: decide, documenta y sé consistente.

Nunca rellenes nulos numéricos con 0 sin pensarlo. Un 0 en "cantidad" significa "no se vendió nada" — que no es lo mismo que "no tenemos el dato". Si tu KPI es el promedio de ventas por transacción, ese 0 falso te baja la media.

### Paso 3: corregir tipos

Pandas lee todo como texto si no le dices lo contrario. Una "fecha" que es un string no se puede filtrar por rango, y una "cantidad" que contiene "abc" no se puede sumar. Aquí es donde conviertes cada columna al tipo que le corresponde.

1# --- TRANSFORM: paso 3 — tipos ---
2# Fechas: to_datetime con errors='coerce' convierte basura en NaT
3df['fecha'] = pd.to_datetime(df['fecha'], errors='coerce')
4
5# Numéricos: to_numeric con errors='coerce' convierte "abc" en NaN
6df['cantidad'] = pd.to_numeric(df['cantidad'], errors='coerce')
7df['precio_unitario'] = pd.to_numeric(df['precio_unitario'], errors='coerce')
8
9# Limpiar lo que haya quedado como NaN tras la conversión
10df = df.dropna(subset=['fecha', 'cantidad', 'precio_unitario'])
11
12# Cantidad debería ser entero
13df['cantidad'] = df['cantidad'].astype(int)
14
15print(df.dtypes)
16print(f"Filas finales: {len(df)}")

errors="coerce" es tu mejor amigo: lo que no puede convertir, lo marca como nulo en vez de explotar.

### Paso 4: transformar y enriquecer

Ya tienes los datos limpios. Ahora puedes añadir columnas calculadas que aporten valor al análisis. Esto es la T de ETL en su versión más útil: no solo limpiar, sino dejar los datos listos para que alguien los consuma sin tener que recalcular.

1# --- TRANSFORM: paso 4 — enriquecer ---
2df['importe_total'] = df['cantidad'] * df['precio_unitario']
3df['mes'] = df['fecha'].dt.to_period('M')
4df['dia_semana'] = df['fecha'].dt.day_name()
5
6print(df[['fecha', 'producto', 'importe_total', 'mes']].head())

Columnas derivadas: el analista no debería tener que multiplicar cantidad × precio cada vez.

### Paso 5: guardar el resultado (Load)

El último paso: guardar los datos limpios en un formato eficiente. Parquet es el estándar de facto en el mundo de datos: comprime bien, mantiene los tipos y lo lee cualquier herramienta (Spark, DuckDB, BigQuery, Pandas). Para un caso simple, un CSV limpio también vale.

1# --- LOAD ---
2# Opción A: Parquet (recomendado para pipelines reales)
3df.to_parquet('ventas_limpias.parquet', index=False)
4
5# Opción B: CSV limpio (si el destino no soporta Parquet)
6df.to_csv('ventas_limpias.csv', index=False)
7
8print(f"Pipeline completado: {len(df)} filas guardadas")
9print(f"Columnas: {list(df.columns)}")

Parquet > CSV para cualquier cosa seria. Pero CSV sirve si tu destino es Excel o un legacy system.

Si dudas: Parquet para pipelines, CSV para compartir con gente que usa Excel.

### El pipeline completo: todo junto

1"""Pipeline de limpieza de ventas — versión completa."""
2import pandas as pd
3
4# --- EXTRACT ---
5df = pd.read_csv('ventas_2024.csv')
6print(f"[E] Leídas {len(df)} filas")
7
8# --- TRANSFORM ---
9# 1. Duplicados
10df = df.drop_duplicates()
11
12# 2. Tipos
13df['fecha'] = pd.to_datetime(df['fecha'], errors='coerce')
14df['cantidad'] = pd.to_numeric(df['cantidad'], errors='coerce')
15df['precio_unitario'] = pd.to_numeric(df['precio_unitario'], errors='coerce')
16
17# 3. Nulos (post-conversión)
18df = df.dropna(subset=['fecha', 'cantidad', 'precio_unitario'])
19df['tienda'] = df['tienda'].fillna('online')
20df['cantidad'] = df['cantidad'].astype(int)
21
22# 4. Enriquecer
23df['importe_total'] = df['cantidad'] * df['precio_unitario']
24df['mes'] = df['fecha'].dt.to_period('M').astype(str)
25
26print(f"[T] {len(df)} filas tras limpieza")
27
28# --- LOAD ---
29df.to_parquet('ventas_limpias.parquet', index=False)
30print(f"[L] Guardado ventas_limpias.parquet ✓")

50 líneas. Esto es un pipeline. Todo lo demás (Airflow, Prefect, crons) son formas de ejecutar esto de forma fiable.

### Qué viene después

Este pipeline funciona, pero tiene limitaciones obvias: si el fichero cambia de formato, se rompe; si falla a medias, no sabes por dónde iba; si lo ejecutas dos veces, duplicas la carga. Los frameworks de orquestación (Airflow, Prefect, Dagster) resuelven exactamente esos problemas: reintentos, logs, planificación y alertas.

Pero el concepto base es siempre el mismo: Extract → Transform → Load. Lo que acabas de hacer es lo que hace un ingeniero de datos todos los días, solo que con más tablas, más fuentes y más reglas de validación. Si quieres practicar esto en profundidad con datasets reales y ejercicios progresivos, lo trabajamos en la plataforma en las asignaturas de Python y de pipelines.

Consejo final: antes de meter un framework, asegúrate de que tu lógica de transformación funciona en un script plano. Si no funciona sin Airflow, no va a funcionar con Airflow — solo vas a tardar más en saber por qué falla.