Ingeniería de Datos Aplicada

Taller ETL: Persistencia Robusta

Hasta ahora, tus datos vivían en la memoria RAM (Pandas), desapareciendo al apagar el equipo. Hoy construiremos el puente hacia la persistencia. Aprenderemos a mover datos de Python a SQL Server de forma segura, rápida y transaccional.

Conceptos: ¿Qué es SQL Server?

Arquitectura de Hoy

graph LR subgraph "Tu Máquina (Host)" Python[Python Script
Pandas + SQLAlchemy] CSV[Archivos CSV] end subgraph "Docker Container" SQL[("SQL Server 2022
Port: 1433")] end CSV --> Python Python -- "ODBC Driver 17" --> SQL style SQL fill:#003366,color:white style Python fill:#FFCC00,color:black

2. El Motor de Conexión

Usaremos SQLAlchemy, el estándar de la industria en Python. Actúa como un traductor (ORM) que permite escribir código Python y ejecutar SQL optimizado por debajo.

db_connection.py


import sqlalchemy as sa
from sqlalchemy import create_engine
import urllib

# 1. Definir parámetros (Docker Container)
params = urllib.parse.quote_plus(
    "DRIVER={ODBC Driver 17 for SQL Server};"
    "SERVER=localhost,1433;"  # Coma, no dos puntos
    "DATABASE=master;"        # DB inicial
    "UID=sa;"                 # Usuario System Admin
    "PWD=TuPasswordFuerte123;" # El que definiste en el docker run
    "TrustServerCertificate=yes;"
)

# 2. Crear el "Engine" (El objeto que mantiene el pool de conexiones)
# fast_executemany=True es VITAL para rendimiento en cargas masivas
connection_string = f"mssql+pyodbc:///?odbc_connect={params}"
engine = create_engine(connection_string, fast_executemany=True)

# 3. Test de conexión
with engine.connect() as connection:
    result = connection.execute(sa.text("SELECT @@VERSION"))
    print(f"Conectado a: {result.fetchone()[0]}")

Ingeniería de Performance: fast_executemany

Por defecto, Python inserta fila por fila (INSERT INTO... repetido 1 millón de veces). Esto es lentísimo. Al activar fast_executemany=True, SQLAlchemy empaqueta los datos en binario y los envía en un solo paquete de red.

~10 min

Sin optimización (100k filas)

~15 seg

Con fast_executemany

3. Transacciones y Atomicidad

Imagina que cargas un archivo de 10,000 transacciones bancarias. El sistema falla en la fila 5,000. ¿Qué pasa con las primeras 4,999?

Sin Transacción: Quedan guardadas. Si corres el script de nuevo, se duplican. El dinero "se crea" de la nada. Caos total.
Con Transacción (ACID): Si una falla, TODAS se reversan (Rollback). La base de datos queda limpia, como si nada hubiera pasado.

sequenceDiagram participant Py as Python Script participant DB as SQL Server Py->>DB: BEGIN TRANSACTION Py->>DB: Insertar Fila 1... OK Py->>DB: Insertar Fila 2... OK Py->>DB: Insertar Fila 3... ERROR! Note over Py,DB: ¡Pánico! ¿Qué hacemos? Py->>DB: ROLLBACK (Deshacer todo) DB-->>Py: Estado restaurado al inicio

atomic_load.py


import pandas as pd
from sqlalchemy import text

def cargar_datos_seguro(df, nombre_tabla, engine):
    # Context Manager de SQLAlchemy maneja la transacción automáticamente
    # Si el bloque termina bien -> COMMIT
    # Si hay error -> ROLLBACK
    try:
        with engine.begin() as connection:
            # 1. Limpiar tabla stage (Opcional)
            connection.execute(text(f"TRUNCATE TABLE {nombre_tabla}"))
            
            # 2. Carga masiva
            # chunksize es importante para no saturar la memoria
            df.to_sql(
                nombre_tabla, 
                con=connection, 
                if_exists='append', 
                index=False,
                chunksize=10000 
            )
        print("✅ Carga Exitosa (COMMIT realizado)")
        
    except Exception as e:
        print("❌ Error detectado. Se ejecutó ROLLBACK automático.")
        print(f"Detalle: {e}")

Reto de Ingeniería #2

The Atomic Loader

Hemos recibido un archivo crítico: ventas_historicas.csv (500,000 filas). Sin embargo, hay una "bomba lógica": la fila 250,000 tiene texto en una columna que debería ser numérica.

Tu Misión

Crear un script que intente cargar el archivo a SQL Server.
El script debe fallar al encontrar el error (Type Error).
CRÍTICO: Al verificar la base de datos después del fallo, debe haber 0 filas. Si encuentras 249,999 filas, el reto está reprobado (Integridad comprometida).

Recursos del Reto

Descargar Dataset Corrupto

Pista: No uses pandas.to_sql por defecto sin una conexión transaccional manejada por SQLAlchemy. Pandas por defecto hace commit por lotes (chunks), lo que podría dejar datos sucios.

Estrategia de Apoyo con IA

¿Problemas con los drivers ODBC? ¿Error de conexión al Docker? Usa nuestro asistente entrenado.

Prompts de Ingeniería para NotebookLM:

"Genera un diagrama Mermaid que explique la diferencia entre engine.connect() y engine.begin() en SQLAlchemy 2.0."
"Tengo un error 'Login failed for user sa'. Revisa mi connection string y compáralo con los logs comunes de Docker MSSQL."
"¿Cómo configuro fast_executemany en Windows vs Linux/Mac? ¿Necesito instalar drivers adicionales?"

Consultar al Asistente