Optimizador de Procesamiento por Lotes (Windags Skills)

Optimizador de Procesamiento por Lotes

Optimiza cargas de trabajo de procesamiento por lotes usando Spark, Polars, DuckDB y pandas, con enfoque en eficiencia de memoria, paralelismo y reducción de costos.

Disparadores de Activación

Activar en: "procesamiento por lotes", "optimización de Spark", "Polars", "DuckDB", "rendimiento de pandas", "data frame", "optimización de shuffle", "asimetría de partición", "optimización de memoria", "fuera de memoria"

NO para: Streaming en tiempo real → streaming-pipeline-architect | Ajuste de consultas en almacenes de datos → data-warehouse-optimizer | Orquestación de pipelines → airflow-dag-orchestrator

Inicio Rápido

1. Elige la herramienta adecuada — DuckDB para análisis en un solo nodo, Polars para DataFrames, Spark para distribuido
2. Perfila primero — identifica cuellos de botella (shuffle, asimetría, memoria) antes de optimizar
3. Reduce datos temprano — filtra y selecciona columnas lo antes posible en el pipeline
4. Evita shuffles — transmite tablas pequeñas, pre-particiona datos, usa joins del lado del mapa
5. Dimensiona recursos correctamente — ajusta la memoria/núcleos del executor al tamaño real de los datos

Capacidades Principales

Dominio	Tecnologías
Distribuido	Apache Spark 3.5+, Dask, Ray
Un solo nodo	DuckDB 1.1+, Polars 1.x, pandas 2.2+
Formatos de archivo	Parquet, Arrow IPC, Delta Lake, Iceberg
Optimización	AQE (Spark), evaluación perezosa (Polars), escaneos columnar
Nube	Databricks, EMR, Dataproc, Spark sin servidor

Patrones Arquitectónicos

Árbol de Decisión para Selección de Herramientas

¿Tamaño de los datos?
  ├─ < 10 GB     → DuckDB (SQL) o Polars (DataFrame)
  │                 Una sola máquina, sin configuración, iteración más rápida
  │
  ├─ 10-100 GB   → Polars (perezoso) o DuckDB (out-of-core)
  │                 Sigue siendo una sola máquina con volcado a disco
  │
  └─ > 100 GB    → Spark (distribuido)
                    Clúster multi-nodo, joins basados en shuffle

¿Complejidad?
  ├─ Centrado en SQL  → DuckDB (motor SQL más rápido para análisis)
  ├─ DataFrame        → Polars (10x más rápido que pandas, evaluación perezosa)
  └─ ML complejo      → Spark + MLlib o Spark + Ray

Patrones de Optimización en Spark

from pyspark.sql import SparkSession
import pyspark.sql.functions as F

spark = SparkSession.builder \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .getOrCreate()

# BIEN: transmitir tabla de dimensiones pequeña (< 100MB)
from pyspark.sql.functions import broadcast
resultado = df_grande.join(broadcast(df_dim_pequena), "clave")

# BIEN: pushdown de predicados — filtrar antes del join
pedidos = spark.read.parquet("s3://data/pedidos/") \
    .filter(F.col("fecha_pedido") >= "2026-01-01") \
    .select("pedido_id", "cliente_id", "monto")  # poda de columnas

# MAL: collect() en conjunto grande de datos — causa OOM en el driver
# todos_datos = df_grande.collect()  # NUNCA hagas esto

# BIEN: escribir salida particionada
resultado.repartition(200) \
    .write.mode("overwrite") \
    .partitionBy("fecha_pedido") \
    .parquet("s3://output/resultados/")

Evaluación Perezosa en Polars

import polars as pl

# Modo perezoso: construye plan de consulta, optimiza y luego ejecuta
resultado = (
    pl.scan_parquet("data/pedidos/*.parquet")  # escaneo perezoso
    .filter(pl.col("fecha_pedido") >= "2026-01-01")
    .join(
        pl.scan_parquet("data/clientes/*.parquet"),
        on="cliente_id",
        how="inner"
    )
    .group_by("region")
    .agg([
        pl.col("monto").sum().alias("ingreso_total"),
        pl.col("pedido_id").n_unique().alias("cantidad_pedidos"),
    ])
    .sort("ingreso_total", descending=True)
    .collect()  # ejecuta el plan optimizado
)

# Polars optimiza: pushdown de predicados, pushdown de proyecciones,
# reordenamiento de joins — todo automáticamente mediante evaluación perezosa

Anti-patrones

1. pandas para >5GB — pandas carga todo en memoria; usa Polars (perezoso) o DuckDB para datos medianos, Spark para grandes
2. Recolectar en el driver — df.collect() o df.toPandas() en DataFrames grandes de Spark causa OOM; agrega primero
3. Ignorar asimetría de particiones — una partición con 10x más datos que otras cuellos de botella en todo el trabajo; usa AQE o salting
4. Leer todas las columnas — siempre selecciona solo las columnas necesarias; el formato columnar Parquet omite columnas no utilizadas por completo
5. Archivos de salida minúsculos — demasiados archivos de salida pequeños (< 128MB) ralentizan las lecturas posteriores; fusiona antes de escribir

Lista de Verificación de Calidad

• Herramienta coincide con el tamaño de los datos (DuckDB/Polars < 100GB, Spark > 100GB)
• Columnas podadas temprano (seleccionar solo lo necesario)
• Filtros empujados al nivel de escaneo (pushdown de predicados)
• Tablas pequeñas transmitidas en joins (< 100MB)
• AQE de Spark habilitado (ejecución adaptativa de consultas)
• Sin collect() en conjuntos de datos grandes (agregar antes de recolectar)
• Archivos de salida de 128MB-1GB (fusionar/reparticionar antes de escribir)
• Asimetría de particiones monitoreada y mitigada (salting o AQE)
• Trabajo perfilado: etapas de Spark UI, .explain() de Polars, EXPLAIN ANALYZE de DuckDB
• Memoria dimensionada apropiadamente: memoria del executor >= 2x la partición más grande