Ingeniería de Datos

Python Core: Robustez y Eficiencia

En este capítulo, dejamos atrás el "scripting básico" para adoptar prácticas de ingeniería de software. Aprenderemos a escribir código Python que no solo funcione, sino que sea resiliente a fallos, eficiente en el procesamiento de datos y capaz de interactuar con el sistema operativo de manera profesional.

2. Lógica Funcional en Python

Python es un lenguaje multiparadigma. Aunque es fuertemente orientado a objetos, en el mundo de los datos (especialmente en Spark y Pandas), utilizamos intensivamente el paradigma funcional. Esto implica evitar estados mutables y preferir transformaciones de datos limpias y predecibles.

2.1. List Comprehensions y Generadores

Las comprensiones de listas son la forma "Pythonica" de transformar datos. Son más rápidas que los bucles for tradicionales porque se ejecutan a nivel de C.

Enfoque Imperativo (Clásico)


precios_crudos = ["$100", "$25.50", "$9.99", "Error"]
precios_limpios = []

for precio in precios_crudos:
    if precio != "Error":
        valor = float(precio.replace("$", ""))
        precios_limpios.append(valor)

Enfoque Funcional (Pythonico)


precios_crudos = ["$100", "$25.50", "$9.99", "Error"]

precios_limpios = [
    float(p.replace("$", "")) 
    for p in precios_crudos 
    if p != "Error"
]

Nota de Ingeniería: Si el dataset es masivo (millones de registros), usaríamos expresiones generadoras (cambiando [] por ()) para no cargar todo en memoria RAM, procesando uno a uno (Lazy Evaluation).

2.2. Funciones Lambda, Map y Filter

Estas funciones son la base conceptual de Apache Spark. Entenderlas en Python puro es prerrequisito para entender RDDs y DataFrames distribuidos.

Transformación (Map)

Aplica una función a cada elemento de una colección.


archivos = ["data_2023.csv", "data_2024.csv", "config.json"]

rutas_completas = list(map(lambda x: f"/mnt/raw/{x}", archivos))

Filtrado (Filter)

Selecciona elementos que cumplen una condición booleana.


archivos = ["data_2023.csv", "data_2024.csv", "config.json", "temp.tmp"]

solo_csvs = list(filter(lambda x: x.endswith(".csv"), archivos))

3. Manejo de Excepciones y Robustez

En Ingeniería de Datos, la pregunta no es "si fallará", sino "cuándo fallará". Archivos corruptos, desconexiones de red y formatos inesperados son el día a día. Un script profesional no se detiene ante un error; lo captura, lo registra (log) y continúa o falla de manera controlada (Graceful Shutdown).

3.1. Anatomía de un Bloque Try-Except Robusto


import logging

logging.basicConfig(
    filename='pipeline.log', 
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def procesar_archivo(ruta_archivo):
    try:
        logging.info(f"Iniciando proceso: {ruta_archivo}")
        
        with open(ruta_archivo, 'r') as f:
            contenido = f.read()
            
        if not contenido:
            raise ValueError("El archivo está vacío")
            
        datos = int(contenido) 
        return datos

    except FileNotFoundError:
        logging.error(f"Error Crítico: El archivo {ruta_archivo} no existe.")
        return None

    except ValueError as e:
        logging.warning(f"Error de Datos en {ruta_archivo}: {e}")
        return 0

    except Exception as e:
        logging.critical(f"Error inesperado: {e}")
        raise 

    finally:
        logging.info(f"Finalizado intento para {ruta_archivo}")

Puntos Clave

Logging vs Print: Nunca uses print en producción. El logging permite auditoría y niveles de severidad.
Especificidad: Captura excepciones específicas (`FileNotFoundError`) antes que las genéricas (`Exception`).
Raise Customizado: Usamos `raise ValueError` para validar reglas de negocio (archivo vacío) proactivamente.
Finally: Se ejecuta siempre. Ideal para cerrar conexiones a bases de datos o liberar memoria.

4. I/O y Sistema de Archivos Moderno

Legacy Alert: En versiones antiguas de Python se usaba mucho os.path. En Python 3.4+ y en ingeniería moderna, el estándar es pathlib. Es orientado a objetos, agnóstico del sistema operativo (funciona igual en Windows/Linux) y más legible.

Navegación y Búsqueda (Globbing)

Buscar patrones de archivos es la tarea #1 en ingesta de datos.


from pathlib import Path

carpeta_raw = Path("./datos/landing")

# Crear carpeta si no existe (mkdir -p)
carpeta_raw.mkdir(parents=True, exist_ok=True)

# Buscar todos los CSVs recursivamente
archivos_csv = list(carpeta_raw.rglob("*.csv"))

for archivo in archivos_csv:
    print(f"Encontrado: {archivo.name}")
    print(f"Ruta Absoluta: {archivo.resolve()}")

Manipulación y Context Managers

Mover, renombrar y leer de forma segura.


from pathlib import Path
import shutil

origen = Path("datos/landing/ventas_2024.csv")
destino_ok = Path("datos/bronze/ventas_2024.csv")
destino_err = Path("datos/error/ventas_2024.csv")

try:
    # Context Manager: Cierra el archivo automáticamente
    with open(origen, 'r', encoding='utf-8') as f:
        header = f.readline()
        if "fecha,monto" not in header:
            raise ValueError("Header inválido")
            
    # Mover archivo (Atomic operation ideally)
    shutil.move(str(origen), str(destino_ok))
    
except Exception as e:
    shutil.move(str(origen), str(destino_err))

Trabajo Asincrónico

Reto de Ingeniería: The Bronze Ingestor

Vas a construir tu primer Pipeline de Ingesta. Tu misión es crear un script de Python robusto que simule un proceso de recepción de datos empresariales.

Clonar Repositorio del Reto

* Incluye dataset de prueba y scripts de validación.

Escenario

Tienes una carpeta /landing donde llegan archivos diarios. Algunos vienen corruptos (vacíos o formato incorrecto). Necesitas automatizar su clasificación.

Carpeta Origen: landing/
Carpeta Éxito: bronze/
Carpeta Fallo: bad_data/

Requerimientos Técnicos

Usar pathlib para escanear archivos.
Implementar try/except. Si un archivo no se puede leer, el script no debe detenerse.
Validación: Un archivo es válido solo si tiene más de 0 bytes.
Mover los archivos a su destino correspondiente usando shutil.

Estrategia de Apoyo con IA (NotebookLM)

No estás solo en este reto. Hemos preparado un entorno de conocimiento en Google NotebookLM cargado con la documentación de pathlib, shutil y las mejores prácticas de manejo de errores en Python.

Cómo usar la IA para resolver el reto:

Ingresa al cuaderno oficial del curso usando el botón de abajo.
Prompt Sugerido 1: "Explícame cómo usar pathlib para iterar sobre todos los archivos .txt de una carpeta y cómo verificar su tamaño en bytes".
Prompt Sugerido 2: "¿Cuál es la diferencia entre usar os.rename y shutil.move para mover archivos entre carpetas en Python? Dame un ejemplo con try-except".
Análisis de Código: Si tu script falla, pega el error en NotebookLM y pregunta: "¿Por qué este bloque try-except no está capturando el error de permiso denegado?".

Acceder al Cuaderno de Soporte