Procesar datos en lotes es habitual en APIs, bases de datos, colas, archivos y pipelines. En lugar de cargar toda la colección o escribir loops con índices, itertools.batched() agrupa cualquier iterable en tuplas del tamaño elegido y consume la entrada de forma perezosa. El último lote puede ser más pequeño o puede rechazarse con strict=True en las versiones que ofrecen esa opción.
Esta guía explica cómo dividir secuencias y generadores, manejar grupos incompletos, enviar peticiones por bloques, leer archivos, combinar lotes con ejecutores, controlar memoria y backpressure, diseñar reintentos y evitar errores comunes.
Primer ejemplo
from itertools import batched
numeros = range(10)
for lote in batched(numeros, 3):
print(lote)
El resultado contiene tuplas de hasta tres valores: (0, 1, 2), (3, 4, 5), (6, 7, 8) y (9,). El iterable no se convierte en una lista completa.
Consumo perezoso
def generar():
for numero in range(1_000_000):
yield numero
primero = next(batched(generar(), 100))
Solo se consumen los primeros cien valores para formar el primer lote. Esto permite trabajar con streams grandes o potencialmente infinitos, siempre que el consumidor también avance de manera controlada.
El último lote incompleto
Por defecto, el grupo final puede contener menos de n elementos. Esta semántica es adecuada cuando todos los datos deben procesarse aunque la cantidad total no sea múltiplo del tamaño.
list(batched([1, 2, 3, 4, 5], 2))
# [(1, 2), (3, 4), (5,)]
Modo strict
for lote in batched(datos, 4, strict=True):
procesar(lote)
Con strict=True, un grupo final incompleto genera ValueError. Es útil para coordenadas, frames, matrices o protocolos que exigen tamaño exacto. Comprueba la versión mínima porque el parámetro se incorporó después de la función inicial.
Tamaños inválidos
El tamaño debe ser al menos uno. Cero y valores negativos son inválidos. Valida configuraciones procedentes de variables de entorno, CLI o servicios remotos antes de crear el iterador.
Por qué slicing no es equivalente
lotes = [datos[i:i + 100] for i in range(0, len(datos), 100)]
El slicing funciona solo con secuencias indexables y crea todos los grupos inmediatamente. No sirve para archivos, generadores, cursores o iteradores arbitrarios. batched() acepta cualquier iterable y produce una tupla cada vez.
Implementación conceptual
from itertools import islice
def batched_manual(iterable, n):
iterator = iter(iterable)
while lote := tuple(islice(iterator, n)):
yield lote
La función oficial sigue esta idea: crea un iterador, extrae hasta n valores con islice() y termina cuando no quedan elementos. Prefiere la API estándar porque comunica intención y resuelve casos límite.
Peticiones a una API
for lote in batched(ids, 100):
respuesta = cliente.buscar_muchos(list(lote))
guardar(respuesta)
Las APIs suelen limitar IDs por petición. El tamaño también debe considerar bytes del payload, rate limits, timeout, tamaño de respuesta y coste de reintento. Cien IDs no siempre significan cien cargas equivalentes.
Inserciones en base de datos
for lote in batched(registros, 500):
cursor.executemany(sql, lote)
conexion.commit()
Los lotes reducen overhead, pero transacciones enormes mantienen locks durante más tiempo y hacen más costoso el rollback. Mide en la base real y elige límites que respeten la atomicidad requerida.
Leer líneas de archivo
with open("eventos.log", encoding="utf-8") as archivo:
for lineas in batched(archivo, 1000):
procesar_lineas(lineas)
Un archivo de texto ya es un iterador de líneas. En cada paso solo es necesario conservar el lote actual. Las strings normalmente incluyen el salto de línea.
Datos binarios
Iterar directamente sobre bytes produce enteros. Para bloques binarios, archivo.read(tamaño) suele ser más eficiente. batched es apropiado cuando la fuente ya entrega unidades lógicas como registros, tokens o mensajes decodificados.
Transformar cada lote
for lote in batched(datos, 50):
normalizados = [normalizar(item) for item in lote]
escribir(normalizados)
Cada lote es una tupla. Conviértelo en lista únicamente cuando una API exija mutabilidad o un array JSON.
Combinar con ejecutores
from concurrent.futures import ThreadPoolExecutor
from itertools import batched
with ThreadPoolExecutor(max_workers=4) as executor:
for resultados in executor.map(procesar_lote, batched(datos, 100)):
guardar(resultados)
Esta composición envía grupos a workers, pero todavía debes considerar prefetch, propagación de excepciones, orden y presión sobre servicios externos. El artículo de concurrent.futures en Python explica los ejecutores.
Iteradores asíncronos
itertools.batched() trabaja con iterables síncronos. No consume directamente un async iterator. Para streams asíncronos, crea un helper:
async def async_batched(source, n):
lote = []
async for item in source:
lote.append(item)
if len(lote) == n:
yield tuple(lote)
lote.clear()
if lote:
yield tuple(lote)
Añade validación de tamaño y comportamiento strict cuando el contrato lo exija.
Backpressure
La evaluación perezosa limita el consumo de la fuente, pero no garantiza backpressure en todo el pipeline. Si cada lote se coloca inmediatamente en una cola sin límite, la memoria puede seguir creciendo. Usa colas limitadas, semáforos o procesamiento secuencial.
Elegir el tamaño
El tamaño ideal depende del coste fijo por llamada, memoria por elemento, objetivos de latencia, límites externos y coste de fallos. Grupos pequeños aumentan overhead; grupos grandes aumentan memoria, locks, latencia y efecto de retries. Mide cargas representativas.
Reintentos e idempotencia
Al repetir un lote, las operaciones deben ser idempotentes o usar claves de idempotencia. De lo contrario, elementos procesados antes de un timeout pueden duplicarse. Si el servicio informa fallos individuales, registra resultados por elemento o divide el grupo problemático.
No agrupa por clave
batched corta el flujo por posición. No reúne registros de la misma categoría. Usa itertools.groupby() para valores adyacentes con una clave común y ordena previamente cuando sea necesario.
No es una ventana deslizante
batched(valores, 2) produce pares no solapados: (a,b), (c,d). pairwise(valores) produce pares solapados: (a,b), (b,c), (c,d). Elige según el algoritmo.
Iteradores de una sola pasada
La fuente original se consume. No esperes recorrerla otra vez. Recrea la fuente o materializa solo cuando el reuso sea necesario. Ten cuidado con tee(), porque consumidores desequilibrados crean un buffer oculto.
Errores dentro de un lote
Define si un elemento inválido rechaza todo el grupo o se envía a una cola de fallos. Operaciones financieras o de inventario pueden requerir rollback atómico. Importaciones analíticas pueden procesar válidos y registrar errores por separado.
Observabilidad
Registra número de lote, cantidad, duración, intentos y tipo de error sin publicar contenido sensible. Métricas de throughput y latencia permiten ajustar el tamaño con datos.
Compatibilidad
Para versiones anteriores, implementa el recipe con islice o utiliza una biblioteca de compatibilidad. Centraliza el fallback para que el resto del proyecto use una única interfaz.
Errores comunes
- Materializar todos los lotes: elimina el beneficio de memoria.
- Ignorar el grupo final corto: usa strict cuando el tamaño exacto importa.
- Aceptar tamaño cero: valida configuración.
- Confundir lotes con ventanas: batched no solapa elementos.
- Alimentar una cola ilimitada: la memoria puede crecer.
- Reintentar operaciones no idempotentes: los efectos pueden duplicarse.
Ejemplo completo: importación resiliente
from itertools import batched
import time
def importar(registros, tamaño=200):
for indice, lote in enumerate(batched(registros, tamaño), 1):
for intento in range(1, 4):
try:
insertar_lote(lote)
registrar_metrica("lote_ok", len(lote))
break
except ErrorTemporal:
if intento == 3:
enviar_a_fallos(indice, lote)
raise
time.sleep(2 ** (intento - 1))
El ejemplo limita memoria, registra progreso y usa backoff exponencial. En producción, añade idempotencia, transacciones, jitter, cancelación y diagnósticos estructurados.
Conclusión
itertools.batched() ofrece una forma clara y perezosa de consumir iterables en bloques. Funciona con listas, generadores, archivos, cursores y otros streams, haciendo explícita la política del último lote.
La documentación oficial de itertools.batched define la API. Elige tamaños a partir de métricas, controla backpressure y usa strict cuando los grupos incompletos sean inválidos.
