Processar dados em lotes é uma necessidade comum em APIs, bancos de dados, filas, arquivos e pipelines. Em vez de carregar toda a coleção ou escrever loops manuais com índices, itertools.batched() agrupa qualquer iterável em tuplas de tamanho definido, consumindo os dados de forma preguiçosa. O último lote pode ser menor, ou pode ser rejeitado com strict=True em versões que oferecem esse parâmetro.
Neste guia, você aprenderá a dividir sequências e geradores, trabalhar com lotes incompletos, enviar requisições em blocos, ler arquivos, combinar batched com executores, controlar memória e evitar erros comuns em processamento incremental.
Primeiro exemplo
from itertools import batched
numeros = range(10)
for lote in batched(numeros, 3):
print(lote)
O resultado contém tuplas de até três elementos: (0, 1, 2), (3, 4, 5), (6, 7, 8) e (9,). O iterável não é convertido em lista inteira.
Consumo preguiçoso
def gerar():
for numero in range(1_000_000):
yield numero
primeiro = next(batched(gerar(), 100))
Apenas os cem primeiros valores são consumidos para formar o primeiro lote. Isso permite trabalhar com streams grandes ou potencialmente infinitos, desde que o consumidor também avance de maneira controlada.
O último lote
Por padrão, o último grupo pode ter menos de n elementos. Essa semântica é adequada quando todos os dados precisam ser processados, mesmo que a quantidade total não seja múltipla do tamanho escolhido.
list(batched([1, 2, 3, 4, 5], 2))
# [(1, 2), (3, 4), (5,)]
Modo strict
for lote in batched(dados, 4, strict=True):
processar(lote)
Com strict=True, um lote final incompleto gera ValueError. Isso é útil quando os registros formam quadros, coordenadas, matrizes ou protocolos que exigem tamanho exato. Verifique a versão mínima de Python, pois o parâmetro foi incorporado depois da função básica.
Tamanho inválido
O tamanho precisa ser pelo menos um. Valores zero ou negativos geram erro. Valide configurações externas antes de criar o iterador.
if tamanho < 1:
raise ValueError("tamanho deve ser positivo")
Por que não usar slicing
lotes = [dados[i:i + 100] for i in range(0, len(dados), 100)]
O slicing funciona em sequências indexáveis, mas não em geradores, arquivos ou iteradores. Também cria uma lista com todos os lotes. batched() aceita qualquer iterável e entrega um grupo por vez.
Implementação conceitual
from itertools import islice
def batched_manual(iterable, n):
iterator = iter(iterable)
while batch := tuple(islice(iterator, n)):
yield batch
A implementação real segue essa ideia: converte a fonte em iterador, usa islice() para retirar até n elementos e encerra quando não há valores. Prefira a função oficial, que documenta a intenção e reduz código.
Enviar dados para uma API
from itertools import batched
for lote in batched(ids, 100):
resposta = cliente.buscar_em_lote(list(lote))
salvar(resposta)
APIs frequentemente limitam a quantidade de IDs por requisição. O lote deve respeitar o limite e também considerar tamanho do payload, timeout, retries e rate limits. Não escolha o número apenas pela memória local.
Inserções em banco
for lote in batched(registros, 500):
cursor.executemany(sql, lote)
conexao.commit()
Commits em lote reduzem overhead, mas grupos grandes prolongam locks e aumentam o custo de rollback. Meça no banco real e use transações coerentes com a atomicidade necessária.
Lendo linhas de arquivo
with open("eventos.log", encoding="utf-8") as arquivo:
for linhas in batched(arquivo, 1000):
processar_linhas(linhas)
O arquivo já é um iterador de linhas. Em cada ciclo, apenas o lote atual permanece necessário. Lembre que cada string inclui a quebra de linha, salvo na última linha.
Lotes de bytes
Iterar diretamente sobre bytes produz inteiros. Se você precisa de blocos binários, ler com arquivo.read(tamanho) costuma ser mais eficiente. batched é apropriado quando a fonte já produz unidades lógicas, como registros ou tokens.
Transformação por lote
for lote in batched(dados, 50):
normalizados = [normalizar(item) for item in lote]
gravar(normalizados)
O lote é uma tupla. Transforme-o em lista apenas quando uma API exigir mutabilidade ou formato JSON.
Paralelismo com Executor
from concurrent.futures import ThreadPoolExecutor
from itertools import batched
with ThreadPoolExecutor(max_workers=4) as executor:
for resultados in executor.map(processar_lote, batched(dados, 100)):
salvar(resultados)
Essa composição limita o número de tarefas ao fluxo do executor, mas ainda é preciso considerar prefetch, exceções, ordem e pressão sobre serviços externos. O guia de concurrent.futures no Python explica executores e futures.
Asyncio
itertools.batched() trabalha com iteráveis síncronos. Um async iterator não pode ser passado diretamente. Para streams assíncronos, escreva um helper com async for que acumule itens e produza lotes, ou use uma biblioteca específica.
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)
Backpressure
A preguiça de batched ajuda, mas não garante backpressure em todo o pipeline. Se o consumidor coloca cada lote em uma fila sem limite, a memória ainda cresce. Use filas limitadas, semáforos ou processamento sequencial para controlar a produção.
Escolhendo o tamanho
O tamanho ideal depende do custo fixo por chamada, memória por registro, latência, limites externos e tolerância a falhas. Lotes pequenos aumentam overhead; lotes grandes aumentam latência, memória e impacto de retries. Faça benchmarks com dados representativos.
Retries
Ao repetir um lote após falha, operações devem ser idempotentes ou usar chaves de idempotência. Caso contrário, registros já processados podem ser duplicados. Para identificar falhas individuais, talvez seja necessário dividir novamente o lote ou registrar resultados por item.
Ordenação e agrupamento
batched apenas corta o fluxo por posição. Ele não agrupa por valor ou chave. Para agrupar registros consecutivos por categoria, use itertools.groupby(). Para janelas sobrepostas, use uma técnica de sliding window, não batched.
Diferença para pairwise
pairwise() produz pares sobrepostos: (a,b), (b,c), (c,d). batched(..., 2) produz pares não sobrepostos: (a,b), (c,d). Escolha de acordo com o problema.
Geradores de uma passagem
O iterador original é consumido. Não espere percorrê-lo novamente depois. Se precisar de reuso, armazene os dados ou crie uma nova fonte. Evite tee() em streams muito desequilibrados, pois ele pode manter um buffer grande.
Erros dentro do lote
Decida se um item inválido cancela todo o lote ou é registrado separadamente. Em sistemas financeiros ou de inventário, atomicidade pode exigir rollback total. Em analytics, talvez seja melhor continuar e enviar falhas para uma fila de rejeitados.
Observabilidade
Registre número do lote, quantidade, duração, tentativas e erros, mas não publique dados sensíveis. Métricas de latência por lote e itens por segundo ajudam a ajustar o tamanho.
Compatibilidade
Em versões anteriores à disponibilidade de itertools.batched, implemente o recipe com islice ou use uma biblioteca compatível. Centralize o fallback para que o restante do projeto use uma única interface.
Erros comuns
- Converter tudo em lista: perde o benefício preguiçoso.
- Ignorar o lote final menor: use strict quando o tamanho exato for obrigatório.
- Usar tamanho zero: valide configurações.
- Confundir lotes com janelas: batched não sobrepõe elementos.
- Enviar lotes sem limite para uma fila: a memória pode crescer.
- Repetir operações não idempotentes: retries podem duplicar efeitos.
Exemplo completo: importação resiliente
from itertools import batched
import time
def importar(registros, tamanho=200):
for indice, lote in enumerate(batched(registros, tamanho), 1):
for tentativa in range(1, 4):
try:
inserir_lote(lote)
registrar_metrica("lote_ok", len(lote))
break
except ErroTemporario:
if tentativa == 3:
enviar_para_falhas(indice, lote)
raise
time.sleep(2 ** (tentativa - 1))
O exemplo limita memória, registra progresso e aplica backoff. Em produção, acrescente idempotência, transação, jitter e cancelamento.
Conclusão
itertools.batched() oferece uma forma clara e preguiçosa de consumir iteráveis em blocos. Ele funciona com listas, geradores e arquivos, reduz boilerplate e torna explícita a política de tamanho e lote final.
A documentação oficial de itertools.batched define a API. Escolha tamanhos com métricas, controle backpressure e use strict=True quando grupos incompletos representarem erro.
