Módulo 4: Escalabilidad y Ejecución Perezosa (Lazy Evaluation)¶

En el mundo real, los ingenieros de datos rara vez trabajan con datasets que quepan cómodamente en la memoria RAM de su laptop. El cuello de botella más común en el procesamiento de datos es el Límite de RAM (RAM Limits).

El Problema: Carga Inmediata (Eager Loading)¶

La mayoría de los analistas comienzan utilizando un enfoque de Ejecución Inmediata (Eager Execution). En este modelo, cada operación se ejecuta en el momento en que se llama, y el resultado se almacena completamente en la memoria RAM.

¿Por qué falla el enfoque Eager con Big Data?¶

Cuando utilizamos funciones como pd.read_csv() o pl.read_csv(), el sistema intenta cargar todo el archivo en la memoria. Si tienes un archivo de 50GB y una RAM de 16GB, el proceso fallará inmediatamente con un error de OutOfMemory (OOM).

Incluso si el archivo cabe en RAM, el enfoque Eager es ineficiente porque:
1. Carga datos innecesarios: Cargamos todas las columnas y filas, aunque solo necesitemos filtrar un pequeño subconjunto.
2. No hay optimización global: El sistema no sabe qué harás después, por lo que no puede optimizar la secuencia de pasos.

Ejemplo: Comparativa Pandas vs Polars (Eager)¶

Veamos cómo se comporta la carga inmediata. En este ejemplo, cargamos un dataset de vuelos y filtramos solo unos pocos registros.

import pandas as pd
import polars as pl
import time

# Enfoque Pandas (Eager)
start = time.time()
df_pd = pd.read_csv('flights.csv', dtype={"TAIL_NUMBER": "string", "ORIGIN_AIRPORT": "string", "DESTINATION_AIRPORT": "string"})
df_pd = df_pd[(df_pd['MONTH'] == 12) & (df_pd['ORIGIN_AIRPORT'] == 'SEA') & (df_pd['DESTINATION_AIRPORT'] == 'DFW')]
print(f"Pandas Eager: {time.time() - start:.4f}s")

# Enfoque Polars (Eager)
start = time.time()
df_pl = pl.read_csv('flights.csv').filter(
    (pl.col('MONTH') == 12) & (pl.col('ORIGIN_AIRPORT') == 'SEA') & (pl.col('DESTINATION_AIRPORT') == 'DFW')
)
print(f"Polars Eager: {time.time() - start:.4f}s")

Aunque Polars es significativamente más rápido que Pandas, ambos siguen cargando todo el CSV en RAM antes de filtrar. Si el CSV fuera más grande que la RAM, ambos fallarían.

La Solución: Ejecución Perezosa (Lazy Evaluation)¶

La Ejecución Perezosa (Lazy Evaluation) cambia el paradigma: en lugar de ejecutar cada línea de código inmediatamente, Polars construye un Plan de Ejecución (Query Plan).

¿Cómo funciona?¶

1. Planificación (.lazy() o scan_*): Cuando usas pl.scan_csv(), Polars no lee los datos. Simplemente anota: "Tengo que leer este archivo, luego filtrar por mes, luego seleccionar estas columnas".
2. Optimización: El optimizador de Polars analiza el plan y aplica reglas para hacerlo más eficiente (por ejemplo, moviendo el filtro lo más cerca posible de la lectura del archivo).
3. Ejecución (.collect()): Solo cuando llamas a .collect(), Polars ejecuta el plan optimizado, leyendo únicamente los datos estrictamente necesarios.

Ejemplo: Polars Lazy¶

# Enfoque Polars (Lazy)
start = time.time()
# scan_csv NO carga los datos en RAM, crea un LazyFrame
lazy_plan = pl.scan_csv('flights.csv').filter(
    (pl.col('MONTH') == 12) & (pl.col('ORIGIN_AIRPORT') == 'SEA') & (pl.col('DESTINATION_AIRPORT') == 'DFW')
)

# .collect() dispara la ejecución optimizada
df_lazy = lazy_plan.collect()
print(f"Polars Lazy: {time.time() - start:.4f}s")

Enfoque en Apache Arrow y Formato Parquet¶

Para que la ejecución perezosa sea verdaderamente potente, necesitamos un formato de almacenamiento que soporte el acceso selectivo. Aquí es donde entra Apache Arrow.

Apache Arrow: El estándar de memoria¶

Apache Arrow es un formato de memoria columnar diseñado para el análisis de datos eficiente. Permite que diferentes herramientas (Polars, DuckDB, Spark) compartan datos sin necesidad de copiarlos (Zero-copy).

Parquet: Arrow en el disco¶

Si Arrow es el formato en memoria, Apache Parquet es su representación en disco. Parquet almacena los datos en columnas en lugar de filas. Esto es fundamental para dos optimizaciones críticas:

1. Projection Pushdown (Poda de Columnas)¶

En un CSV, para leer la columna 10, el sistema debe leer las primeras 9. En Parquet, Polars puede saltar directamente a la columna necesaria en el disco y leer solo esa.
* Resultado: Menos I/O de disco y menor uso de RAM.

2. Predicate Pushdown (Poda de Filas)¶

Parquet almacena metadatos (como el valor mínimo y máximo) para bloques de datos. Si filtras MONTH == 12, Polars puede leer los metadatos y decir: "Este bloque de 10,000 filas solo tiene meses del 1 al 6, no necesito leerlo".
* Resultado: El sistema ignora archivos o bloques enteros de datos antes de cargarlos en RAM.

Ejemplo: Optimización con Parquet¶

# Planificar acceso con Predicate y Projection Pushdown
lf = pl.scan_parquet("events/*.parquet")
result = (
    lf.filter(pl.col("country") == "IN")               # Predicate Pushdown
      .filter(pl.col("ts") >= pl.datetime(2025, 9, 1)) # Predicate Pushdown
      .select(["ts", "country", "path"])              # Projection Pushdown
      .collect()
)

En este código, Polars no lee todo el directorio events/. Lee los metadatos de Parquet, descarta los archivos que no cumplen los filtros y solo extrae las tres columnas solicitadas.

Casos Prácticos de Escalabilidad¶

1. Datasets Particionados (Hive Partitioning)¶

Si los datos están organizados en carpetas como año=2025/mes=10/, Polars puede optimizar la búsqueda basándose únicamente en la estructura de carpetas.

# Búsqueda focalizada en particiones específicas
lf = pl.scan_parquet("path/YYYY=2025/MM=10/*/*.parquet")
today = lf.filter(pl.col("DD") == "08").collect()

2. Joins Eficientes entre Formatos¶

La ejecución perezosa permite unir datos de diferentes fuentes (CSV y Parquet) optimizando la operación global antes de cargar cualquier dato en RAM.

q_join = (
    pl.scan_parquet("Parquet/*/*/*.parquet")
    .join(pl.scan_csv("taxi_zone_lookup.csv"), left_on="PULocationID", right_on="LocationID")
    .filter(pl.col("total_amount") > 25)
    .group_by("Zone")
    .agg(
        (pl.col("total_amount") /
        (pl.col("tpep_dropoff_datetime") - pl.col("tpep_pickup_datetime")).dt.total_minutes()
        ).mean().alias("cost_per_minute")
    ).sort("cost_per_minute", descending=True)
)

r_join = q_join.collect()

3. Pipeline de Conversión Directa (Streaming)¶

Puedes convertir datos de CSV a Parquet sin cargar el archivo completo en RAM utilizando .sink_parquet(), que procesa los datos en batches (streaming).

lf = pl.scan_csv("raw/*.csv", dtypes={"ts": pl.Datetime, "user_id": pl.Int64})
(
    lf.filter(pl.col("ts") >= pl.datetime(2025, 10, 1))
      .with_columns(pl.col("path").str.slice(0, 64).alias("path_64"))
      .sink_parquet("curated/2025-10.parquet")
)

Resumen: Eager vs Lazy¶