Patrones de Procesamiento por Lotes

Comienza con 100-1000 items por batch. Haz benchmark con tus datos y restricciones de memoria. Para inserts en base de datos, 500-2000 rows por batch balancea throughput y tamaño de transacción. Para procesamiento de archivos, 10-50 MB por chunk evita OOM en contenedores de 512 MB. Monitorea el uso de memoria con resource.getrusage(resource.RUSAGE_SELF).ru_maxrss en Python. Si los batches exceden la memoria disponible, reduce el batch size o cambia a streaming. Para APIs network-bound, batches más grandes (500-5000) amortizan latencia. Para transforms CPU-bound, batches más pequeños (100-500) permiten mejor paralelismo. Siempre testea con volúmenes de datos similares a producción — los datos sintéticos raramente revelan patrones de presión de memoria.

Usa Celery/Redis para sistemas distribuidos con múltiples workers, retry logic y monitoreo. Celery provee task routing, priority queues y dead-letter handling out of the box. Para pipelines simples de un solo nodo, cron jobs bastan — pero añade flock para prevenir runs overlapping: flock -n /tmp/batch.lock python batch_job.py. Para Kubernetes, usa recursos CronJob con concurrencyPolicy: Forbid. Consulta Rate Limiting para controlar throughput. Para setups cloud-native, AWS Step Functions o Google Cloud Workflows proveen retry y state management manejados sin overhead de infraestructura.

Versiona tu lógica de trabajo y schemas de datos. Ejecuta versiones viejas y nuevas en paralelo durante la migración. Usa una columna schema_version en las tablas target: ALTER TABLE orders ADD COLUMN schema_version INT DEFAULT 1. El batch job verifica schema_version y aplica la transformación correcta. Para Avro/Protobuf, usa schema registry para manejar cambios backward-compatible. Para JSON, valida con JSON Schema versionado por $id. Despliega nuevas versiones del job con un feature flag: if config.use_new_schema: transform_v2(record) else: transform_v1(record). Monitorea la salida de ambas versiones para discrepancias. Una vez confiado, descomisiona la versión vieja.

El checkpointing guarda el progreso para que los jobs fallidos resuman desde el último batch exitoso. Almacena checkpoints en una tabla durable: CREATE TABLE job_checkpoints (job_id VARCHAR(64), batch_number INT, status VARCHAR(20), PRIMARY KEY (job_id, batch_number)). Después de cada batch, escribe: INSERT INTO job_checkpoints VALUES ('daily_etl', 42, 'completed'). En restart, query el último batch completado: SELECT MAX(batch_number) FROM job_checkpoints WHERE job_id = 'daily_etl' AND status = 'completed'. Resume desde el batch 43. Para checkpoints basados en archivos, escribe un archivo JSON a S3 o disco local después de cada batch. Usa writes atómicos para prevenir corrupción: escribe a un temp file, luego rename. Para jobs distribuidos, usa Redis o etcd para coordinación distribuida de checkpoints.

Las dead-letter queues (DLQ) aislan items fallidos para inspección manual y retry. En Python, mantén una DLQ list: dead_letter_queue = [] y appendea items fallidos con contexto: dead_letter_queue.append({'item': item, 'error': str(e), 'batch_id': batch_id, 'timestamp': datetime.utcnow()}). Después de que el job completa, escribe la DLQ a una tabla de base de datos o message queue. En Celery, configura task_queues con un DLQ exchange dedicado. En AWS SQS, setea RedrivePolicy para mover mensajes después de maxReceiveCount intentos. Para DLQs basadas en base de datos: CREATE TABLE dead_letters (id SERIAL PRIMARY KEY, job_id VARCHAR(64), item JSONB, error TEXT, created_at TIMESTAMP DEFAULT NOW()). Procesa items de DLQ separadamente con un retry job que corre cada hora.

Emite métricas en cada boundary de batch. En Python, usa prometheus_client: from prometheus_client import Counter, Histogram; processed = Counter('batch_processed_total', 'Items processed'); latency = Histogram('batch_duration_seconds', 'Batch duration'). Push métricas a Prometheus Pushgateway para batch jobs: from prometheus_client import push_to_gateway; push_to_gateway('localhost:9091', job='batch_etl', registry=registry). Para setups cloud, emite CloudWatch custom metrics o Datadog statsd. Trackea: items procesados, items fallidos, duración del batch, queue depth y uso de memoria. Setea dashboards de Grafana con alerts para: throughput abajo del 50% del average, failure rate arriba del 5% y duración del job excediendo SLA. Loguea JSON estructurado para cada batch: {"batch_id": 42, "processed": 1000, "failed": 3, "duration_ms": 1250}.

El backpressure ocurre cuando los sistemas downstream no pueden mantener el ritmo del productor de batches. Implementa rate limiting con un semaphore: from threading import Semaphore; rate_limiter = Semaphore(10) — acquire antes de cada batch write y release después. Para writes a base de datos, usa connection pooling con un max pool size para limitar inserts concurrentes. Para API calls, implementa un token bucket: import time; time.sleep(1 / max_rps). Monitorea la latencia downstream — si aumenta más de 2x el baseline, reduce el batch size o pausa el procesamiento. En Celery, setea worker_prefetch_multiplier = 1 para prevenir que los workers over-fetchen. Para pipelines basados en Kafka, setea max.poll.records para controlar el batch size. Usa concurrent.futures.ThreadPoolExecutor(max_workers=N) para limitar paralelismo en Python.

Testea batch jobs con data fixtures determinísticos. Crea un dataset de test: test_data = [{'id': i, 'value': i * 2} for i in range(10000)]. Testea idempotencia corriendo el job dos veces y verificando que la salida sea idéntica: assert run_job(test_data) == run_job(test_data). Testea fault tolerance inyectando fallos: def failing_handler(batch): if len(batch) > 500: raise Exception('simulated failure') y verifica que el job retiene y registra fallos. Testea checkpointing matando el job mid-run y verificando que resume correctamente. Usa property-based testing con Hypothesis: @given(st.lists(st.dictionaries(st.text(), st.integers()))) para generar inputs edge-case. Para jobs basados en SQL, usa testcontainers para levantar una base de datos real: from testcontainers.postgres import PostgresContainer.

Los batch jobs de series temporales procesan datos en ventanas de tiempo fijas. Usa windowed batching: agrupa records por window_start y window_end timestamps. Por ejemplo, SELECT date_trunc('hour', timestamp) AS window, COUNT(*) FROM events GROUP BY 1 procesa batches por hora. Para datos que llegan tarde, usa watermarking: permite datos hasta 5 minutos tarde seteando watermark = current_time - 5 minutes. Procesa ventanas solo después de que el watermark pase. Para retención, particiona por tiempo: CREATE TABLE events_2025_01 PARTITION OF events FOR VALUES FROM ('2025-01-01') TO ('2025-02-01'). Dropea particiones viejas en lugar de borrar rows: DROP TABLE events_2024_01. Para downsampling, agrega datos crudos en tablas resumen de 1-minuto, 1-hora y 1-día en batch jobs separados.

El procesamiento exactly-once requiere writes idempotentes y checkpoints transaccionales. Usa una transacción para escribir tanto la salida del batch como el checkpoint atómicamente: BEGIN; INSERT INTO results SELECT * FROM staging; INSERT INTO job_checkpoints VALUES ('job_1', 42, 'completed'); COMMIT;. Si la transacción falla, tanto los datos como el checkpoint se rollbackean — el job retiene el batch. Para consumers de Kafka, usa enable.idempotence=true y transactional.id para exactly-once del lado del producer. Para writes a base de datos, usa INSERT ... ON CONFLICT DO NOTHING para manejar batches duplicados safe. Para calls a APIs externas, usa idempotency keys: Idempotency-Key: batch_42_run_3 en headers de request. Acepta que exactly-once tiene un coste de performance — mide si at-least-once con writes idempotentes basta para tu use case.