Patrones de Procesamiento por Lotes

Visión General

La mayoría de pipelines de datos, trabajos ETL y reportes nocturnos siguen funcionando por lotes. Trabajar sobre chunks acotados en lugar de un stream interminable mantiene la lógica fácil de razonar — a cambio hay que pensar con detalle en idempotencia, tolerancia a fallos y observabilidad.

Cuándo Usar

Recurre a estos patrones cuando:

• Procesas datasets grandes que no caben en memoria. Consulta Retry Logic para manejar fallos transitorios.
• Construyes pipelines ETL para data warehouses
• Generas reportes o agregaciones nocturnas
• Migras datos entre sistemas con ventanas de mantenimiento

Descarta el procesamiento por lotes cuando los eventos necesitan latencia inferior al minuto, el dataset cabe holgadamente en memoria o el trabajo llega como un stream continuo: para esos casos encaja mejor un framework de streaming o una cola de mensajes.

Solución

Pipeline Resiliente de Procesamiento por Lotes (Python)

import logging
from typing import Callable, List, Iterator

class BatchProcessor:
    def __init__(self, batch_size: int = 1000, max_retries: int = 3):
        self.batch_size = batch_size
        self.max_retries = max_retries
        self.processed = 0
        self.failed = []

    def process(
        self,
        items: Iterator[dict],
        handler: Callable[[List[dict]], None]
    ) -> dict:
        batch = []
        for item in items:
            batch.append(item)
            if len(batch) >= self.batch_size:
                self._execute(batch, handler)
                batch = []

        if batch:
            self._execute(batch, handler)

        return {"processed": self.processed, "failed": len(self.failed)}

    def _execute(self, batch: List[dict], handler: Callable):
        for attempt in range(self.max_retries):
            try:
                handler(batch)
                self.processed += len(batch)
                return
            except Exception as e:
                logging.warning(f"Batch fallido (intento {attempt + 1}): {e}")
                if attempt == self.max_retries - 1:
                    self.failed.extend(batch)

Seguimiento Idempotente de Trabajos (SQL)

CREATE TABLE job_runs (
    job_id VARCHAR(64) PRIMARY KEY,
    started_at TIMESTAMP NOT NULL DEFAULT NOW(),
    completed_at TIMESTAMP,
    status VARCHAR(20) CHECK (status IN ('running', 'completed', 'failed')),
    checksum VARCHAR(64)
);

-- Antes de comenzar, verifica si ya está completado
SELECT * FROM job_runs WHERE job_id = 'daily_report_2025_01_15' AND status = 'completed';

Explicación

Antes de confiar un pipeline por lotes a producción, asegúrate de que cumple tres propiedades:

1. Idempotencia: el mismo trabajo, ejecutado dos veces, tiene que llegar al mismo resultado. Usa IDs de trabajo y checksums para saltarte el trabajo ya procesado. Consulta Endpoints Idempotentes para patrones de deduplicación.
2. Tolerancia a fallos: un lote malo no debería tumbar todo el trabajo. Implementa reintentos con backoff exponencial y una dead-letter queue.
3. Observabilidad: necesitas ver progreso, throughput y errores mientras corre el trabajo, así que emite métricas de ítems procesados, latencia y tasa de fallos — la receta de logging estructurado cubre el formato de log.

Estrategia de chunking: el tamaño del lote es un equilibrio entre memoria y throughput — demasiado pequeño y el overhead domina, demasiado grande y coquetea con el OOM.

Variantes

Patrón	Caso de uso	Compromiso
Procesamiento por chunks	Archivos grandes, límites de memoria	Más simple, mayor latencia
Workers paralelos	Transformaciones CPU-bound	Complejo, necesita coordinación
MapReduce	Agregación distribuida	Escala horizontalmente
Change Data Capture	Sincronización incremental	Requiere soporte de la fuente

Tamaño de los Lotes y Control del Backpressure

Empieza con 100-1000 ítems por lote y haz benchmarks con datos reales. Para inserts en base de datos, 500-2000 filas por lote balancean throughput y tamaño de transacción; para procesamiento de archivos, 10-50 MB por chunk evita OOM en contenedores de 512 MB. Mide la memoria con resource.getrusage(resource.RUSAGE_SELF).ru_maxrss en Python. Las APIs limitadas por red toleran lotes más grandes (500-5000) que amortizan la latencia por request, mientras que las transformaciones CPU-bound rinden mejor con lotes pequeños (100-500) que se paralelizan bien entre workers.

Cuando el sistema de destino no da abasto, regula al productor en lugar de dejar crecer la cola. Un semáforo limita las escrituras de lotes en vuelo:

from threading import Semaphore

rate_limiter = Semaphore(10)  # máximo 10 escrituras de lote concurrentes

El tamaño del pool de conexiones ya acota la concurrencia de escritura en base de datos. En Celery, usa worker_prefetch_multiplier = 1 para que los workers no descarguen trabajo de más; en consumers de Kafka, ajusta max.poll.records. Vigila la latencia del destino: cuando supere aproximadamente 2 veces la línea base, reduce el tamaño del lote o pausa el productor.

Particionado y data skew

Divide los trabajos grandes por una clave de partición para que los workers corran en paralelo: por fecha (df['created_at'].dt.strftime('%Y-%m-%d')), por hash (df['id'] % num_partitions) o por rango (WHERE id BETWEEN 0 AND 99999). Los layouts estilo Hive (data/year=2025/month=01/day=15/data.parquet) funcionan con Spark, Presto y Athena, y el particionado declarativo de PostgreSQL (CREATE TABLE orders PARTITION BY RANGE (created_at)) te da la misma división dentro de la base de datos.

El skew aparece cuando una partición eclipsa a las demás: compruébalo con Counter(df['partition_key']).most_common(10); si la mayor es 10 veces la mediana, tienes un straggler. Añade sal a la clave caliente (key + '_' + str(random.randint(0, 9))), procesa las subparticiones de forma independiente y combina los resultados al final.

Fiabilidad: Reintentos, Checkpoints y Dead Letters

Reintentos. Vuelve a intentar primero el lote completo; baja al reintento por ítem solo para los registros que siguen fallando. El backoff exponencial con jitter evita que todos los reintentos lleguen a la vez, y tenacity resuelve ambas cosas limpiamente:

import random
from tenacity import retry, stop_after_attempt, wait_exponential, wait_random

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=4, max=60) + wait_random(0, 2)
)
def process_batch(batch):
    ...

En Celery, usa autoretry_for = (ConnectionError,) con retry_backoff = True en la tarea. Registra cada reintento con el ID del lote para que el equipo pueda distinguir ruido transitorio de una dependencia que se está cayendo.

Checkpoints. Un trabajo de seis horas que falla a las 5:55 debería reanudar, no reiniciar. Persiste el progreso tras cada lote:

CREATE TABLE job_checkpoints (
    job_id VARCHAR(64),
    batch_number INT,
    status VARCHAR(20),
    PRIMARY KEY (job_id, batch_number)
);

Al reiniciar, ejecuta SELECT MAX(batch_number) FROM job_checkpoints WHERE job_id = 'daily_etl' AND status = 'completed' y reanuda desde el siguiente lote. Los checkpoints en archivo también funcionan: escribe un JSON en S3 o en disco local tras cada lote y usa renombrado atómico (escribe en un archivo temporal y luego os.replace) para que un fallo a mitad de escritura no corrompa el checkpoint. Los trabajos distribuidos pueden coordinar checkpoints con Redis o etcd.

Dead letters. Aísla los ítems que fallan de forma permanente para inspeccionarlos, en lugar de descartarlos o tumbar la ejecución:

CREATE TABLE dead_letters (
    id SERIAL PRIMARY KEY,
    job_id VARCHAR(64),
    item JSONB,
    error TEXT,
    created_at TIMESTAMP DEFAULT NOW()
);

Guarda el ítem, el error y el ID del lote, y reprocésalos con un trabajo separado cada hora una vez corregida la causa raíz. AWS SQS lo hace de forma nativa con RedrivePolicy y maxReceiveCount; en Celery, enruta los fallos a un exchange DLQ dedicado.

Cargas Incrementales y Evolución del Esquema

Procesa solo lo que cambió desde la última ejecución guardando una marca de agua (high-water mark):

CREATE TABLE job_watermarks (
    job_id VARCHAR(64) PRIMARY KEY,
    last_processed_id BIGINT,
    last_processed_at TIMESTAMP
);

-- Siguiente ejecución: solo filas nuevas
SELECT * FROM orders
WHERE id > (SELECT last_processed_id FROM job_watermarks WHERE job_id = 'orders_etl')
ORDER BY id LIMIT 10000;

Las marcas basadas en timestamp (updated_at > last_processed_at) sirven cuando las filas se actualizan in situ, pero se pierden los borrados físicos: la replicación lógica los captura (PostgreSQL pgoutput, binlog de MySQL vía Debezium).

Los esquemas también cambian a mitad de pipeline. Añade una columna schema_version (ALTER TABLE orders ADD COLUMN schema_version INT DEFAULT 1), haz que el trabajo bifurque según su valor (transform_v2(record) if record['schema_version'] == 2 else transform_v1(record)) y ejecuta ambas versiones en paralelo hasta verificar la migración. Los equipos con Avro o Protobuf obtienen la misma garantía de un schema registry con evolución retrocompatible; los pipelines JSON pueden versionar esquemas por $id y validar con JSON Schema.

Calidad de Datos, Validación y Reconciliación

Ejecuta la validación antes de que los datos lleguen al destino, no después. Great Expectations cubre las comprobaciones habituales en pocas líneas:

batch.expect_column_values_to_not_be_null("id")
batch.expect_column_values_to_be_unique("id")
batch.expect_column_values_to_be_between("amount", 0, 10000)

Pon en cuarentena las filas que fallen (INSERT INTO quarantine SELECT * FROM staging WHERE amount IS NULL OR amount < 0) y emite un reporte por lote — {"batch_id": 42, "total": 1000, "valid": 995, "invalid": 5} — con una alerta cuando la tasa de inválidos supere el 1%.

Reconcilia la salida contra el origen para detectar pérdidas silenciosas. Compara el conteo de filas por lote y, para datos financieros, las sumas:

SELECT (SELECT COUNT(*) FROM source.orders WHERE date = '2025-01-15')
     - (SELECT COUNT(*) FROM target.orders WHERE date = '2025-01-15') AS discrepancy;

Persiste los resultados en una tabla reconciliation_results y alerta sobre cualquier varianza distinta de cero en cuestión de minutos: un trabajo nocturno que pierde un 2% de filas durante un mes es mucho más difícil de reconstruir que uno detectado la misma noche.

Orquestación y Programación

Un cron job en un solo nodo basta para pipelines simples: añade flock para que ejecuciones solapadas no se corrompan entre sí:

flock -n /tmp/batch.lock python batch_job.py

En Kubernetes, CronJob con concurrencyPolicy: Forbid te da la misma garantía. El trabajo distribuido pertenece a una cola de verdad: Celery/Redis ofrece enrutado de tareas, reintentos y gestión de dead letters, mientras que AWS Step Functions y Google Cloud Workflows proporcionan máquinas de estado gestionadas sin mantener brokers.

Airflow orquesta DAGs de lotes con varios pasos:

with DAG(
    "daily_etl",
    schedule="@daily",
    default_args={"retries": 3, "retry_delay": timedelta(minutes=5)},
) as dag:
    process_task = PythonOperator(task_id="process", python_callable=process_batch)

Define sla para alertas de plazo, usa Pools para limitar la concurrencia entre DAGs y mapea tareas sobre particiones con TaskGroup y expand. En serverless, mantén cada invocación dentro del timeout (15 minutos en Lambda), devuelve ReportBatchItemFailures para que SQS reencole solo los ítems fallidos y encadena los lotes a través de la cola en lugar de ejecutar un único trabajo largo.

Archivos Más Grandes que la Memoria

Procesa los archivos fila a fila en streaming en lugar de cargarlos enteros. csv.DictReader combina bien con more_itertools.chunked:

import csv
from more_itertools import chunked

with open("large.csv") as f:
    for batch in chunked(csv.DictReader(f), 1000):
        process(batch)

Pandas expone el mismo patrón con pd.read_csv("large.csv", chunksize=10000), y Node.js hace streaming con csv-parse. Los archivos binarios o de registros de tamaño fijo piden otra herramienta: mmap mapea el archivo en memoria virtual y el sistema operativo pagina los datos bajo demanda en lugar de cargarlo entero:

import mmap

with open("large.bin", "r+b") as f:
    mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
    for i in range(0, len(mm), 65536):
        process(mm[i:i + 65536])
    mm.close()

Para archivos de más de 10 GB, usa split -l 1000000 large.csv chunk_ y procesa las partes en paralelo entre varios workers.

Lo que funciona

• Diseña para idempotencia: cada trabajo debe poder reintentarse con seguridad
• Registra todo: inicio del trabajo, fin y resultado de cada lote
• Usa transacciones: envuelve las escrituras por lotes en transacciones de base de datos
• Monitoriza la profundidad de cola: alerta cuando los lotes pendientes superen umbrales
• Implementa circuit breakers: deja de reintentar si el destino está caído

Errores Comunes

1. No manejar fallos parciales: un lote de 1000 donde falla 1 necesita reintento individual
2. Ignorar los límites de memoria: cargar datasets enteros en RAM tumba el proceso
3. Falta de checkpointing: un trabajo de 6 horas que falla a las 5:55 debe reiniciar desde cero
4. Pérdida silenciosa de datos: errores registrados en logs pero invisibles para el equipo
5. Sin estrategia de rollback: los trabajos fallidos dejan la base de datos en estado inconsistente

Solución de Problemas

• La salida del pipeline no coincide con lo esperado: valida los esquemas de entrada, los estados intermedios y el conteo de filas en cada paso.
• La calidad de datos se degrada con el tiempo: añade validaciones y detección de anomalías; define SLIs de frescura, completitud y exactitud.
• El trabajo falla de forma intermitente: busca condiciones de carrera, dependencias externas y contención de recursos; reintenta con idempotencia y backoff acotado.
• Los cambios de esquema rompen a los consumidores: usa schema registries y evolución retrocompatible.
• Los costes de almacenamiento crecen sin esperarlo: audita la retención de particiones, la compresión y las copias duplicadas; archiva los datos fríos y define políticas de ciclo de vida.

Lectura Adicional

• Documentación de Apache Airflow — creación de DAGs, reintentos y semántica de programación
• Kafka Connect — conectores gestionados de origen y destino para ingesta por lotes
• Cuotas de AWS Lambda — límites de timeout y tamaño de lote para trabajos serverless
• Particionado de tablas en PostgreSQL — particionado declarativo para tablas orientadas a lotes
• Great Expectations — framework de validación de datos usado más arriba

Ver También

• Ejemplos ejecutables en el repo companion — implementación de BatchProcessor, esquema de seguimiento y suite de pytest
• Guía de Pipelines ETL — arquitectura completa de extracción, transformación y carga
• Runbook de Dead Letter Queue — operación de una DLQ en producción
• Indexación de Bases de Datos — optimización del rendimiento de consultas para lecturas por lotes
• Load Testing — validación del rendimiento de trabajos por lotes bajo carga