La misma limpieza, cada lunes
El dueño manda una exportación nueva cada lunes. Tiene los mismos fallos que la semana pasada: repeticiones, palabras donde van números, escrituras raras.
Podrías limpiarla a mano en un cuaderno cada vez. O escribir los pasos una vez, en orden, en un script que se ejecuta igual siempre.
Al final de esta lección
Nota
Puedo escribir un pequeño pipeline que limpia, une e informa, y que se para cuando falla una comprobación.
Palabras clave
| Palabra | Qué significa | Ejemplo |
|---|---|---|
| Pipeline | pasos ejecutados en orden, cada uno con el resultado del anterior | |
| Etapa | un paso del pipeline | quitar repetidas |
| Comprobación | una prueba dentro del pipeline que puede pararlo | |
| Código de salida | el número con el que termina un script, 0 si fue bien | sys.exit(1) |
| Reproducible | que da el mismo resultado cada vez que se ejecuta |
Las celdas que rellenan los estudiantes quedan en blanco en esta vista previa.
Etapas en un orden fijo
Un pipeline es una lista de etapas. Cada una recibe una tabla y devuelve otra. El orden importa: no puedes unir por la fecha antes de que las fechas estén limpias.
Tras cada etapa, el pipeline puede imprimir cuántas filas quedan. Una caída brusca es la primera señal de que un paso salió mal.
Ejemplo resuelto: las etapas del pipeline de hoy
| Etapa | Qué hace | Filas esperadas |
|---|---|---|
| lectura | Carga la exportación sucia | 367 |
| sin repetidas | Quita las filas que están dos veces | 365 |
| recuentos reales | Conserva alquileres de 0 o más, como números | 363 |
| unión | Añade la columna de festivos conservando cada día | 363 |
Las celdas que rellenan los estudiantes quedan en blanco en esta vista previa.
Escribir de nuevo la lista de festivos
docker exec jupyter sh -c "printf '%s\n' 'date,holiday' '2025-01-01,Año Nuevo' '2025-01-06,Reyes' '2025-04-17,Jueves Santo' '2025-04-18,Viernes Santo' '2025-05-01,Fiesta del Trabajo' '2025-08-15,Asunción' '2025-12-08,Inmaculada' '2025-12-25,Navidad' > holidays.csv"Se muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
El script tiene unas cuarenta líneas
El siguiente bloque muestra el script. Léelo primero. La función stage imprime las filas que quedan y para todo el script con el código 1 si hay menos que least.
El número least se da al ejecutarlo, así que tú eliges lo estricta que es la comprobación.
steps/pipeline.py
import sys
import pandas as pd
from mess import make
least = int(sys.argv[1])
def stage(name, table):
print(name, "->", len(table), "filas")
if len(table) < least:
print("PARADA:", name, "deja menos de", least, "filas")
sys.exit(1)
return table
make()
table = stage("lectura", pd.read_csv("messy.csv"))
table = stage("sin repetidas", table.drop_duplicates())
table["hires"] = pd.to_numeric(table["hires"], errors="coerce")
table = stage("recuentos reales", table[table["hires"] >= 0])
table["weather"] = table["weather"].str.strip().str.lower()
table = stage("unión", table.merge(pd.read_csv("holidays.csv"), on="date", how="left"))
table.to_csv("report.csv", index=False)
print(table.groupby("weather")["hires"].mean().round(1).to_string())
Escribir el pipeline en el laboratorio
docker exec jupyter sh -c "printf '%s\n' 'import sys' 'import pandas as pd' 'from mess import make' '' 'least = int(sys.argv[1])' '' 'def stage(name, table):' ' print(name, \"->\", len(table), \"filas\")' ' if len(table) < least:' ' print(\"PARADA:\", name, \"deja menos de\", least, \"filas\")' ' sys.exit(1)' ' return table' '' 'make()' 'table = stage(\"lectura\", pd.read_csv(\"messy.csv\"))' 'table = stage(\"sin repetidas\", table.drop_duplicates())' 'table[\"hires\"] = pd.to_numeric(table[\"hires\"], errors=\"coerce\")' 'table = stage(\"recuentos reales\", table[table[\"hires\"] >= 0])' 'table[\"weather\"] = table[\"weather\"].str.strip().str.lower()' 'table = stage(\"unión\", table.merge(pd.read_csv(\"holidays.csv\"), on=\"date\", how=\"left\"))' 'table.to_csv(\"report.csv\", index=False)' 'print(table.groupby(\"weather\")[\"hires\"].mean().round(1).to_string())' > steps/pipeline.py"Se muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Ejecutar con un mínimo de 300 filas
docker exec jupyter python3 steps/pipeline.py 300Se muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Una comprobación que para el pipeline
Una comprobación solo sirve si puede decir que no. Aquí es un mínimo de filas que quedan. Dale un mínimo que los datos no puedan cumplir: el pipeline para en la primera etapa que lo rompe.
El script termina entonces con código 1, y quien lo lanzó sabe que falló.
Pedir al menos 365 filas
docker exec jupyter python3 steps/pipeline.py 365 || echo "código de salida $?"Se muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Pedir al menos 400 filas
docker exec jupyter python3 steps/pipeline.py 400 || echo "código de salida $?"Se muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Comprobar lo que escribió la ejecución terminada
docker exec jupyter python3 -c "import pandas as pd; r=pd.read_csv('report.csv'); print(r.shape, r['holiday'].notna().sum(), r['date'].is_unique, r['hires'].isna().sum())"Se muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Un error frecuente
Advertencia
Mucha gente cree que un pipeline que terminó sin error es correcto. Solo pasó las comprobaciones que alguien escribió. Los números de filas no dicen nada de si las palabras del tiempo están escritas igual. Añade una comprobación por cada fallo que ya hayas visto.
El patrón de una etapa, calculado en Python
# El patrón de una etapa en Python simple: hacer, avisar y parar si quedan muy pocas filas.
rows = [3, 8, 1, 9, 4, 2]
least = 3
def keep_over(rows, limit):
kept = []
for x in rows:
if x > limit:
kept.append(x)
return kept
def stage(name, rows, limit):
rows = keep_over(rows, limit)
print(name, "->", len(rows), "filas")
if len(rows) < least:
print("PARADA:", name)
return None
return rows
rows = stage("quedarse con más de 2", rows, 2)
if rows is not None:
rows = stage("quedarse con más de 5", rows, 5)
if rows is None:
print("parado")
else:
print("terminado")
# Cambia least a 2 y vuelve a ejecutar.
Para ir más lejos: añadir una comprobación
Añade a steps/pipeline.py una etapa que pare si alguna palabra del tiempo no es cloudy, rainy o sunny. Usa set(table["weather"]) y compara.
Pruébala con los datos limpios y luego antes de la línea strip. ¿Qué detecta esta comprobación que los números de filas no vieron?