Boas Práticas de Generators e yield em Python: Lazy Evaluation e Pipelines de Dados para Times Ágeis Já leu

Entendendo Generators: O Que Realmente São

Um generator em Python é uma função que produz uma sequência de valores, mas diferentemente de uma função comum que retorna todos os valores de uma vez, um generator entrega esses valores sob demanda. Internamente, um generator mantém seu estado entre chamadas sucessivas, permitindo que você processe grandes volumes de dados sem carregar tudo na memória simultânea.

A forma mais simples de criar um generator é usar a palavra-chave yield dentro de uma função. Quando Python encontra yield, a função não retorna imediatamente — ela pausa sua execução e retorna o valor especificado. Na próxima chamada, a função resume exatamente de onde parou. Isso é fundamentalmente diferente de return, que encerra a função e descarta seu estado local.

# Função comum que retorna uma lista
def numeros_normais():
    resultado = []
    for i in range(5):
        resultado.append(i * 2)
    return resultado

# Generator que produz valores sob demanda
def numeros_generator():
    for i in range(5):
        yield i * 2

# Usando a função comum
print(numeros_normais())  # [0, 2, 4, 6, 8]

# Usando o generator
gen = numeros_generator()
print(next(gen))  # 0
print(next(gen))  # 2
print(next(gen))  # 4

A diferença crucial está no uso de memória. A função comum cria e armazena toda a lista em memória antes de retornar. O generator, por sua vez, calcula cada valor sob demanda e descarta o anterior, ocupando uma fração do espaço de memória.

Lazy Evaluation: Computação Sob Demanda

Lazy evaluation (avaliação preguiçosa) é o princípio central que torna generators poderosos. Em vez de processar dados antecipadamente, você computa apenas o que é necessário, quando é necessário. Isso é especialmente valioso ao trabalhar com dados infinitos, fluxos contínuos ou arquivos gigantes.

Considere um cenário real: você precisa processar um arquivo com 10 gigabytes de linhas. Se você usasse uma abordagem tradicional, teria que carregar o arquivo inteiro na memória antes de processar a primeira linha. Com lazy evaluation, você lê e processa linha por linha, mantendo apenas a atual na memória.

# Abordagem tradicional (problema de memória)
def ler_arquivo_inteiro(caminho):
    linhas = []
    with open(caminho, 'r') as f:
        for linha in f:
            linhas.append(linha.strip())
    return linhas

# Abordagem com lazy evaluation
def ler_arquivo_lazy(caminho):
    with open(caminho, 'r') as f:
        for linha in f:
            yield linha.strip()

# Processando dados
for linha in ler_arquivo_lazy('dados.txt'):
    processar(linha)  # Apenas uma linha na memória por vez

Expressões Geradoras

Python oferece uma sintaxe compacta para criar generators sem usar def e yield. Expressões geradoras funcionam como list comprehensions, mas usam parênteses em vez de colchetes:

# List comprehension (cria lista inteira na memória)
quadrados_lista = [x**2 for x in range(1000000)]

# Expressão geradora (computa sob demanda)
quadrados_gen = (x**2 for x in range(1000000))

print(type(quadrados_lista))  # <class 'list'>
print(type(quadrados_gen))     # <class 'generator'>

# Consumindo o generator
for quad in quadrados_gen:
    if quad > 50:
        print(quad)
        break

Expressões geradoras são ideais quando você precisa processar dados sequencialmente e não necessita acessá-los aleatoriamente (sem usar índices).

Pipelines de Dados com Generators

Um pipeline de dados é uma série de transformações aplicadas sequencialmente a um fluxo de dados. Generators são perfeitos para construir pipelines porque cada estágio processa dados sob demanda, criando um fluxo eficiente que nunca mantém os dados completos na memória.

A arquitetura é simples: cada função geradora consome dados de um gerador anterior e produz dados transformados para o próximo. Essa abordagem separa responsabilidades, facilita testes e otimiza o consumo de memória.

# Estágio 1: Leitura de dados
def ler_numeros(inicio, fim):
    for i in range(inicio, fim):
        print(f"  Lendo {i}")
        yield i

# Estágio 2: Filtrar números pares
def filtrar_pares(sequencia):
    for numero in sequencia:
        if numero % 2 == 0:
            print(f"  Filtrando {numero}")
            yield numero

# Estágio 3: Multiplicar por 2
def multiplicar(sequencia, fator):
    for numero in sequencia:
        resultado = numero * fator
        print(f"  Multiplicando {numero} por {fator} = {resultado}")
        yield resultado

# Construir o pipeline
pipeline = multiplicar(
    filtrar_pares(
        ler_numeros(1, 11)
    ),
    fator=10
)

# Consumir o pipeline
print("\nResultados:")
for resultado in pipeline:
    print(f"Resultado final: {resultado}")

Saída esperada:

Lendo 1
Lendo 2
Filtrando 2
Multiplicando 2 por 10 = 20
Resultado final: 20
Lendo 3
Lendo 4
Filtrando 4
Multiplicando 4 por 10 = 40
Resultado final: 40
[...]

Pipelines Práticos com CSV

Aqui está um exemplo realista de processamento de dados CSV sem carregar tudo na memória:

import csv
from datetime import datetime

# Estágio 1: Ler CSV linha por linha
def ler_csv(caminho):
    with open(caminho, 'r', encoding='utf-8') as f:
        leitor = csv.DictReader(f)
        for linha in leitor:
            yield linha

# Estágio 2: Filtrar registros válidos
def filtrar_validos(linhas):
    for linha in linhas:
        try:
            float(linha['valor'])
            yield linha
        except (ValueError, KeyError):
            continue

# Estágio 3: Enriquecer com data de processamento
def adicionar_timestamp(linhas):
    for linha in linhas:
        linha['data_processamento'] = datetime.now().isoformat()
        yield linha

# Estágio 4: Agrupar por categoria
def agrupar_por_categoria(linhas):
    grupos = {}
    for linha in linhas:
        categoria = linha.get('categoria', 'sem_categoria')
        if categoria not in grupos:
            grupos[categoria] = []
        grupos[categoria].append(linha)

    for categoria, items in grupos.items():
        yield {'categoria': categoria, 'itens': items}

# Executar pipeline
pipeline = agrupar_por_categoria(
    adicionar_timestamp(
        filtrar_validos(
            ler_csv('vendas.csv')
        )
    )
)

# Processar resultado
for grupo in pipeline:
    print(f"Categoria: {grupo['categoria']}, Quantidade: {len(grupo['itens'])}")

Composição com itertools

A biblioteca padrão itertools oferece generators prontos que facilitam composição de pipelines:

import itertools

# Dados de exemplo
dados = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

# Pipeline: pegar 5 primeiros, filtrar pares, multiplicar por 2
resultado = map(
    lambda x: x * 2,
    filter(
        lambda x: x % 2 == 0,
        itertools.islice(dados, 5)
    )
)

print(list(resultado))  # [4, 8]

# Usando itertools.chain para combinar múltiplas sequências
seq1 = [1, 2, 3]
seq2 = [4, 5, 6]
combinado = itertools.chain(seq1, seq2)

for valor in combinado:
    print(valor)  # 1, 2, 3, 4, 5, 6

# itertools.cycle para repetir sequências infinitamente
ciclo = itertools.cycle(['A', 'B', 'C'])
print([next(ciclo) for _ in range(5)])  # ['A', 'B', 'C', 'A', 'B']

Erros Comuns e Armadilhas

Consumindo Generators Múltiplas Vezes

Um erro frequente é tentar iterar sobre o mesmo generator duas vezes. Após ser consumido, um generator é esgotado e não pode ser reutilizado.

gen = (x**2 for x in range(5))

# Primeira iteração funciona
print(list(gen))  # [0, 1, 4, 9, 16]

# Segunda iteração retorna vazio
print(list(gen))  # []

Se você precisa reutilizar os dados, converta para lista (se a memória permitir) ou crie um novo generator:

# Solução 1: Converter para lista (cuidado com memória)
dados = list(gen)
print(list(dados))
print(list(dados))

# Solução 2: Criar função que retorna novo generator
def criar_gen():
    return (x**2 for x in range(5))

print(list(criar_gen()))
print(list(criar_gen()))

StopIteration e next()

Ao usar next() diretamente, você recebe StopIteration quando o generator se esgota. Use a forma segura com valor padrão:

gen = (x for x in range(3))

print(next(gen))      # 0
print(next(gen))      # 1
print(next(gen))      # 2
print(next(gen, None))  # None (ao invés de StopIteration)

Debugging de Generators

Generators podem ser difíceis de debugar porque não computam valores até serem consumidos. Uma técnica útil é usar itertools.tee() para criar cópias independentes:

import itertools

gen = (x**2 for x in range(5))

# Criar duas cópias independentes
gen1, gen2 = itertools.tee(gen, 2)

# Debugar uma cópia sem afetar a outra
print("Debug:", list(gen1))
print("Consumir:", list(gen2))

Conclusão

Os três pontos fundamentais que você aprendeu neste artigo são:

1. Generators são máquinas de estado que mantêm contexto entre chamadas, viabilizando a implementação de lazy evaluation sem necessidade de armazenar dados completos em memória. Isso transforma a forma como você trabalha com dados grandes ou contínuos.

2. Pipelines de dados com generators oferecem composição elegante e eficiente de transformações, onde cada estágio processa sob demanda. Essa abordagem separa responsabilidades, facilita testes unitários e reduz drasticamente o consumo de memória em comparação com abordagens que materializam dados intermediários.

3. Expressões geradoras e a biblioteca itertools fornecem ferramentas prontas que eliminam a necessidade de escrever generators manualmente em muitos casos, tornando o código mais conciso e legível mantendo todos os benefícios de lazy evaluation.

Domine esses conceitos e você terá uma ferramenta poderosa para lidar com processamento de dados em escala, desde leitura de arquivos gigantes até construção de pipelines ETL complexos.

Referências

• Python Official Documentation - Generators
• Real Python - Generators and yield
• David Beazley - Generator Tricks for Systems Programmers
• Fluent Python - Luciano Ramalho (Capítulo sobre Iterables e Generators)
• Python itertools Documentation