Geradores em Python: Guia Completo de yield, Expressões Geradoras e Avaliação Preguiçosa
Updated on
Processar um arquivo de logs de 10GB ou fazer streaming de milhões de registros de banco de dados pode colocar sua aplicação Python de joelhos. A abordagem tradicional de carregar todos os dados na memória de uma vez leva a gargalos de performance, erros de memória e usuários frustrados. É aqui que os geradores em Python se tornam essenciais — eles permitem processar conjuntos de dados massivos com um consumo mínimo de memória, gerando valores sob demanda em vez de armazenar tudo antecipadamente.
O que são Geradores em Python e Por que Eles Importam
Geradores são funções especiais que produzem uma sequência de valores ao longo do tempo, em vez de computar e retornar tudo de uma vez. Diferentemente de funções comuns que usam return para devolver um único resultado, geradores usam a palavra-chave yield para produzir uma série de valores, pausando a execução entre cada valor e retomando quando o próximo valor é solicitado.
A vantagem fundamental dos geradores é a avaliação preguiçosa (lazy evaluation) — valores são gerados apenas quando necessários. Isso oferece dois benefícios críticos:
- Eficiência de memória: geradores não armazenam a sequência inteira na memória. Um gerador que produz um bilhão de números consome a mesma memória que um que produz dez números.
- Performance: o processamento pode começar imediatamente com o primeiro valor produzido por
yield, sem esperar que todo o conjunto de dados seja preparado.
Aqui vai uma comparação simples que ilustra a diferença:
# Traditional approach - loads entire list into memory
def get_squares_list(n):
result = []
for i in range(n):
result.append(i * i)
return result
# Generator approach - produces values one at a time
def get_squares_generator(n):
for i in range(n):
yield i * i
# Memory impact comparison
import sys
# List approach
squares_list = get_squares_list(1000000)
print(f"List memory: {sys.getsizeof(squares_list):,} bytes") # ~8,000,000 bytes
# Generator approach
squares_gen = get_squares_generator(1000000)
print(f"Generator memory: {sys.getsizeof(squares_gen):,} bytes") # ~112 bytesA diferença de memória é impressionante — o gerador usa 99,999% menos memória do que a lista neste exemplo. Essa diferença cresce dramaticamente com conjuntos de dados maiores.
A palavra-chave yield: o Coração das Funções Geradoras
A palavra-chave yield é o que transforma uma função comum em uma função geradora. Quando o Python encontra yield, ele sabe que deve retornar um objeto gerador em vez de executar a função imediatamente.
def countdown(n):
print(f"Starting countdown from {n}")
while n > 0:
yield n
n -= 1
print("Countdown complete!")
# Creating the generator doesn't execute the function
gen = countdown(3)
print(type(gen)) # <class 'generator'>
# Values are produced on-demand
print(next(gen)) # Starting countdown from 3 -> 3
print(next(gen)) # 2
print(next(gen)) # 1
# next(gen) # Countdown complete! -> Raises StopIterationComportamentos-chave para entender:
- A execução pausa em cada
yielde retoma exatamente daquele ponto na próxima chamada - Variáveis locais mantêm seu estado entre chamadas de
yield - A exceção StopIteration é levantada quando a função geradora retorna (fica sem valores)
Vários yield podem aparecer em um único gerador:
def data_pipeline():
# Phase 1: Loading
yield "Loading data..."
# Phase 2: Processing
yield "Processing records..."
# Phase 3: Validation
yield "Validating results..."
# Phase 4: Complete
yield "Pipeline complete!"
for status in data_pipeline():
print(status)Protocolo de Geradores: Entendendo iter() e next()
Geradores implementam o protocolo de iteradores por meio de dois métodos especiais:
__iter__(): retorna o próprio objeto iterador (o gerador)__next__(): retorna o próximo valor do gerador
Isso torna geradores perfeitos para uso em loops for e outros contextos de iteração. Entender esse protocolo ajuda a esclarecer como geradores funcionam por baixo dos panos:
def simple_gen():
yield 1
yield 2
yield 3
gen = simple_gen()
# These are equivalent
print(gen.__next__()) # 1
print(next(gen)) # 2
# for loops call __next__() automatically until StopIteration
for value in simple_gen():
print(value) # 1, 2, 3Você também pode implementar manualmente o protocolo de iteradores para criar um comportamento semelhante ao de geradores:
class CountDown:
def __init__(self, start):
self.current = start
def __iter__(self):
return self
def __next__(self):
if self.current <= 0:
raise StopIteration
self.current -= 1
return self.current + 1
# Behaves like a generator
for num in CountDown(3):
print(num) # 3, 2, 1No entanto, funções geradoras são muito mais concisas e legíveis do que classes iteradoras implementadas manualmente.
Expressões Geradoras vs List Comprehensions
Expressões geradoras oferecem uma sintaxe concisa para criar geradores, semelhante a list comprehensions, mas com parênteses em vez de colchetes:
# List comprehension - creates entire list in memory
squares_list = [x * x for x in range(10)]
print(type(squares_list)) # <class 'list'>
print(squares_list) # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
# Generator expression - creates generator object
squares_gen = (x * x for x in range(10))
print(type(squares_gen)) # <class 'generator'>
print(squares_gen) # <generator object at 0x...>
# Consume the generator
print(list(squares_gen)) # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]Comparação de sintaxe:
| Recurso | List Comprehension | Expressão Geradora |
|---|---|---|
| Sintaxe | [expr for item in iterable] | (expr for item in iterable) |
| Retorna | Objeto list | Objeto generator |
| Memória | Armazena todos os valores | Gera sob demanda |
| Velocidade | Mais rápida para conjuntos pequenos | Mais rápida para conjuntos grandes |
| Reutilizável | Sim (pode iterar várias vezes) | Não (esgota após uma iteração) |
Exemplo prático mostrando a diferença de memória:
import sys
# List comprehension for 1 million numbers
list_comp = [x for x in range(1000000)]
print(f"List comprehension: {sys.getsizeof(list_comp):,} bytes")
# Generator expression for the same range
gen_exp = (x for x in range(1000000))
print(f"Generator expression: {sys.getsizeof(gen_exp):,} bytes")
# Output:
# List comprehension: 8,000,056 bytes
# Generator expression: 112 bytesExpressões geradoras são ideais quando você só precisa iterar pelos valores uma vez e quer minimizar o uso de memória.
yield from: Delegando para Subgeradores
A instrução yield from simplifica a delegação para subgeradores ou outros iteráveis. Em vez de fazer um loop manual e dar yield em cada valor, yield from cuida disso automaticamente:
# Without yield from
def get_numbers_manual():
for i in range(3):
yield i
for i in range(10, 13):
yield i
# With yield from
def get_numbers_delegated():
yield from range(3)
yield from range(10, 13)
print(list(get_numbers_manual())) # [0, 1, 2, 10, 11, 12]
print(list(get_numbers_delegated())) # [0, 1, 2, 10, 11, 12]Isso é particularmente útil para “achatar” estruturas aninhadas:
def flatten(nested_list):
for item in nested_list:
if isinstance(item, list):
yield from flatten(item) # Recursive delegation
else:
yield item
nested = [1, [2, 3, [4, 5]], 6, [7, [8, 9]]]
print(list(flatten(nested))) # [1, 2, 3, 4, 5, 6, 7, 8, 9]yield from também lida corretamente com exceções e valores de retorno de subgeradores, tornando-se essencial para pipelines de geradores mais complexos.
Avançado: Métodos send() e throw()
Geradores podem ser mais do que produtores de valores — eles também podem receber valores e tratar exceções por meio dos métodos send() e throw(), permitindo comunicação bidirecional no estilo corrotina.
Usando send() para Enviar Valores para Dentro de Geradores
def running_average():
total = 0
count = 0
average = None
while True:
value = yield average # Yield current average, receive new value
total += value
count += 1
average = total / count
# Create generator
avg = running_average()
next(avg) # Prime the generator (advance to first yield)
# Send values and receive running averages
print(avg.send(10)) # 10.0
print(avg.send(20)) # 15.0
print(avg.send(30)) # 20.0
print(avg.send(40)) # 25.0O método send() tanto envia um valor para dentro do gerador (que se torna o resultado da expressão yield) quanto avança a execução até o próximo yield.
Usando throw() para Injetar Exceções
def error_handling_gen():
try:
while True:
value = yield
print(f"Received: {value}")
except ValueError as e:
print(f"Caught ValueError: {e}")
yield "Recovered from error"
except GeneratorExit:
print("Generator is closing")
gen = error_handling_gen()
next(gen) # Prime the generator
gen.send(10) # Received: 10
gen.send(20) # Received: 20
result = gen.throw(ValueError, "Invalid value") # Caught ValueError: Invalid value
print(result) # Recovered from error
gen.close() # Generator is closingEsses recursos avançados são especialmente úteis para implementar máquinas de estado, corrotinas e padrões assíncronos complexos.
Geradores Infinitos: Sequências Sem Fim
Geradores se destacam ao produzir sequências infinitas porque nunca precisam materializar a sequência inteira na memória:
# Infinite counter
def count_from(start=0, step=1):
current = start
while True:
yield current
current += step
# Fibonacci sequence
def fibonacci():
a, b = 0, 1
while True:
yield a
a, b = b, a + b
# Cycling through a sequence
def cycle(iterable):
saved = []
for item in iterable:
yield item
saved.append(item)
while saved:
for item in saved:
yield item
# Usage examples
counter = count_from(10, 2)
for _ in range(5):
print(next(counter)) # 10, 12, 14, 16, 18
fib = fibonacci()
print([next(fib) for _ in range(10)]) # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
colors = cycle(['red', 'green', 'blue'])
print([next(colors) for _ in range(8)]) # ['red', 'green', 'blue', 'red', 'green', 'blue', 'red', 'green']Geradores infinitos são particularmente úteis para fluxos de eventos, monitoramento contínuo e padrões de iteração com estado.
Encadeando Geradores: Construindo Pipelines de Processamento de Dados
Um dos padrões mais poderosos com geradores é encadeá-los para criar pipelines eficientes de processamento de dados. Cada etapa processa dados de forma preguiçosa e passa os resultados para a próxima etapa sem armazenar resultados intermediários:
# Stage 1: Read lines from a file (generator)
def read_log_file(filename):
with open(filename, 'r') as f:
for line in f:
yield line.strip()
# Stage 2: Filter lines containing 'ERROR'
def filter_errors(lines):
for line in lines:
if 'ERROR' in line:
yield line
# Stage 3: Extract timestamp and message
def parse_error_lines(lines):
for line in lines:
parts = line.split(' - ')
if len(parts) >= 2:
yield {'timestamp': parts[0], 'message': parts[1]}
# Stage 4: Count errors by hour
def group_by_hour(errors):
from collections import defaultdict
hourly_counts = defaultdict(int)
for error in errors:
hour = error['timestamp'][:13] # Extract hour portion
hourly_counts[hour] += 1
return hourly_counts
# Build pipeline
log_lines = read_log_file('app.log')
error_lines = filter_errors(log_lines)
parsed_errors = parse_error_lines(error_lines)
results = group_by_hour(parsed_errors)
print(results)Esse pipeline processa um arquivo de log potencialmente enorme com uso mínimo de memória — apenas uma linha fica na memória por vez até a etapa final de agregação.
Outro exemplo com transformação de dados:
# Pipeline: numbers -> square -> filter evens -> sum
def square_numbers(numbers):
for n in numbers:
yield n * n
def filter_even(numbers):
for n in numbers:
if n % 2 == 0:
yield n
# Chain the pipeline
numbers = range(1, 11) # 1-10
squared = square_numbers(numbers)
evens = filter_even(squared)
result = sum(evens) # Only even squares
print(result) # 220 (4 + 16 + 36 + 64 + 100)Comparação de Memória: Benchmark de Gerador vs Lista
Vamos fazer um benchmark real de memória e performance para quantificar os benefícios dos geradores:
import sys
import time
import tracemalloc
def process_with_list(n):
"""Traditional approach using lists"""
tracemalloc.start()
start_time = time.time()
# Create list of squares
squares = [x * x for x in range(n)]
# Filter even squares
even_squares = [x for x in squares if x % 2 == 0]
# Sum results
result = sum(even_squares)
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
elapsed = time.time() - start_time
return result, peak / 1024 / 1024, elapsed # Convert to MB
def process_with_generator(n):
"""Generator approach"""
tracemalloc.start()
start_time = time.time()
# Generator pipeline
squares = (x * x for x in range(n))
even_squares = (x for x in squares if x % 2 == 0)
result = sum(even_squares)
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
elapsed = time.time() - start_time
return result, peak / 1024 / 1024, elapsed
# Benchmark with 1 million numbers
n = 1000000
list_result, list_memory, list_time = process_with_list(n)
gen_result, gen_memory, gen_time = process_with_generator(n)
print(f"Results match: {list_result == gen_result}")
print(f"\nList approach:")
print(f" Memory: {list_memory:.2f} MB")
print(f" Time: {list_time:.4f} seconds")
print(f"\nGenerator approach:")
print(f" Memory: {gen_memory:.2f} MB")
print(f" Time: {gen_time:.4f} seconds")
print(f"\nMemory savings: {((list_memory - gen_memory) / list_memory * 100):.1f}%")Saída típica:
Results match: True
List approach:
Memory: 36.21 MB
Time: 0.0892 seconds
Generator approach:
Memory: 0.12 MB
Time: 0.0624 seconds
Memory savings: 99.7%A abordagem com geradores usa 99,7% menos memória e roda 30% mais rápido — uma melhoria dramática que se acumula com conjuntos de dados maiores.
O Módulo itertools: Utilitários para Geradores
O módulo itertools do Python fornece uma coleção de ferramentas poderosas baseadas em geradores para iteração eficiente. Esses utilitários são escritos em C e altamente otimizados:
Funções Essenciais de itertools
import itertools
# chain - concatenate multiple iterables
combined = itertools.chain([1, 2], [3, 4], [5, 6])
print(list(combined)) # [1, 2, 3, 4, 5, 6]
# islice - slice an iterable (like list slicing but for generators)
numbers = itertools.count() # Infinite counter: 0, 1, 2, 3...
first_ten = itertools.islice(numbers, 10)
print(list(first_ten)) # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
# count - infinite counter with start and step
counter = itertools.count(start=10, step=2)
print([next(counter) for _ in range(5)]) # [10, 12, 14, 16, 18]
# cycle - infinite repetition of an iterable
colors = itertools.cycle(['red', 'green', 'blue'])
print([next(colors) for _ in range(7)]) # ['red', 'green', 'blue', 'red', 'green', 'blue', 'red']
# accumulate - cumulative sums or other operations
numbers = [1, 2, 3, 4, 5]
cumulative = itertools.accumulate(numbers)
print(list(cumulative)) # [1, 3, 6, 10, 15]
# accumulate with custom function
import operator
products = itertools.accumulate(numbers, operator.mul)
print(list(products)) # [1, 2, 6, 24, 120]
# groupby - group consecutive elements by key
data = [('A', 1), ('A', 2), ('B', 3), ('B', 4), ('C', 5)]
for key, group in itertools.groupby(data, key=lambda x: x[0]):
print(f"{key}: {list(group)}")
# A: [('A', 1), ('A', 2)]
# B: [('B', 3), ('B', 4)]
# C: [('C', 5)]Combinações Práticas com itertools
# Paginating results with islice
def paginate(iterable, page_size):
iterator = iter(iterable)
while True:
page = list(itertools.islice(iterator, page_size))
if not page:
break
yield page
# Usage
data = range(25)
for page_num, page in enumerate(paginate(data, 10), 1):
print(f"Page {page_num}: {page}")
# Page 1: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
# Page 2: [10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
# Page 3: [20, 21, 22, 23, 24]
# Windowed iteration (sliding window)
def window(iterable, size):
it = iter(iterable)
win = list(itertools.islice(it, size))
if len(win) == size:
yield tuple(win)
for item in it:
win = win[1:] + [item]
yield tuple(win)
print(list(window([1, 2, 3, 4, 5], 3)))
# [(1, 2, 3), (2, 3, 4), (3, 4, 5)]Casos de Uso do Mundo Real
Lendo Arquivos Grandes Linha a Linha
def process_large_csv(filename):
"""Process a multi-GB CSV file efficiently"""
with open(filename, 'r') as f:
# Skip header
next(f)
for line in f:
# Parse and yield record
fields = line.strip().split(',')
yield {
'user_id': fields[0],
'action': fields[1],
'timestamp': fields[2]
}
# Process millions of records with minimal memory
for record in process_large_csv('user_events.csv'):
# Process one record at a time
if record['action'] == 'purchase':
print(f"Purchase by user {record['user_id']}")Processamento de Dados em Streaming
def stream_api_data(url, batch_size=100):
"""Stream paginated API data without loading all results"""
offset = 0
while True:
response = requests.get(url, params={'offset': offset, 'limit': batch_size})
data = response.json()
if not data:
break
for item in data:
yield item
offset += batch_size
# Process unlimited API results
for item in stream_api_data('https://api.example.com/records'):
process_item(item)Iteração de Resultados de Consulta ao Banco de Dados
def fetch_users_batch(cursor, batch_size=1000):
"""Fetch database records in batches without loading all into memory"""
while True:
results = cursor.fetchmany(batch_size)
if not results:
break
for row in results:
yield row
# Database query
cursor.execute("SELECT * FROM users WHERE active = 1")
# Process millions of users efficiently
for user in fetch_users_batch(cursor):
send_email(user['email'], generate_report(user))Exemplo de Pipeline de ETL
# Extract: Read from source
def extract_from_csv(filename):
with open(filename, 'r') as f:
for line in f:
yield line.strip().split(',')
# Transform: Clean and convert data
def transform_records(records):
for record in records:
yield {
'id': int(record[0]),
'name': record[1].title(),
'email': record[2].lower(),
'age': int(record[3]) if record[3] else None
}
# Load: Write to database
def load_to_database(records, db_connection):
for record in records:
db_connection.execute(
"INSERT INTO users VALUES (?, ?, ?, ?)",
(record['id'], record['name'], record['email'], record['age'])
)
yield record # Pass through for logging
# Build ETL pipeline
raw_data = extract_from_csv('users.csv')
transformed = transform_records(raw_data)
loaded = load_to_database(transformed, db_conn)
# Execute pipeline and count processed records
processed_count = sum(1 for _ in loaded)
print(f"Processed {processed_count} records")Boas Práticas de Geradores e Armadilhas Comuns
Boas Práticas
-
Use expressões geradoras para casos simples
# Simple transformation - use generator expression squares = (x * x for x in range(1000)) # Complex logic - use generator function def complex_processing(data): for item in data: # Multi-step processing result = step1(item) result = step2(result) if validate(result): yield result -
Encadeie geradores para pipelines de dados
# Each stage processes lazily data = read_source() filtered = filter_stage(data) transformed = transform_stage(filtered) results = aggregate_stage(transformed) -
Use
yield frompara delegaçãodef process_all_files(directory): for filename in os.listdir(directory): yield from process_file(filename)
Armadilhas Comuns
-
Geradores são esgotados após uma iteração
gen = (x for x in range(3)) print(list(gen)) # [0, 1, 2] print(list(gen)) # [] - exhausted! # Solution: Convert to list or recreate generator data = list(gen) # If data fits in memory # OR gen = (x for x in range(3)) # Recreate -
Geradores não suportam len() nem indexação
gen = (x for x in range(10)) # len(gen) # TypeError # gen[5] # TypeError # Solution: Convert to list if you need these operations items = list(gen) print(len(items)) print(items[5]) -
Cuidado com escopo e closure em geradores
# Wrong - all generators will use final value of i generators = [lambda: i for i in range(3)] print([g() for g in generators]) # [2, 2, 2] # Correct - capture i in default argument generators = [lambda i=i: i for i in range(3)] print([g() for g in generators]) # [0, 1, 2] -
Tratamento de exceções em cadeias de geradores
def stage1(): for i in range(5): if i == 3: raise ValueError("Error in stage1") yield i def stage2(data): try: for item in data: yield item * 2 except ValueError as e: print(f"Caught: {e}") yield -1 # Error marker # Exception is caught in stage2 for result in stage2(stage1()): print(result)
Comparação: Geradores vs Listas vs Iteradores vs map/filter
| Recurso | Geradores | Listas | Iteradores | map/filter |
|---|---|---|---|---|
| Uso de memória | Mínimo (preguiçoso) | Dataset inteiro | Mínimo (preguiçoso) | Mínimo (preguiçoso) |
| Velocidade de criação | Instantânea | Depende do tamanho | Instantânea | Instantânea |
| Reutilizável | Não | Sim | Não | Não |
| Indexável | Não | Sim | Não | Não |
| Suporte a len() | Não | Sim | Não | Não |
| Modificação | Somente leitura | Mutável | Somente leitura | Somente leitura |
| Sequências infinitas | Sim | Não | Sim | Sim |
| Sintaxe | yield ou () | [] | iter() | map(), filter() |
| Melhor para | Datasets grandes, pipelines | Datasets pequenos, acesso aleatório | Implementação do protocolo | Transformações funcionais |
Exemplo de comparação:
# All produce same results but with different characteristics
data = range(1000000)
# Generator - memory efficient, not reusable
gen = (x * 2 for x in data)
# List - memory intensive, reusable, indexable
lst = [x * 2 for x in data]
# map - memory efficient, functional style
mapped = map(lambda x: x * 2, data)
# Iterator - explicit protocol implementation
class Doubler:
def __init__(self, data):
self.data = iter(data)
def __iter__(self):
return self
def __next__(self):
return next(self.data) * 2
iterator = Doubler(data)Experimentando com Geradores no Jupyter
Ao explorar padrões de geradores e características de performance, trabalhar em um ambiente interativo de notebook acelera o aprendizado. RunCell (opens in a new tab) traz assistência com IA diretamente para notebooks Jupyter, tornando-se ideal para cientistas de dados que estão experimentando pipelines de processamento de dados baseados em geradores.
Com RunCell, você pode:
- Prototipar rapidamente funções geradoras e testar características de memória
- Fazer benchmark de performance de geradores vs listas com datasets reais
- Construir e depurar pipelines complexos de geradores de forma interativa
- Receber sugestões de IA para otimizar workflows de ETL baseados em geradores
Veja como você pode explorar geradores em um notebook:
# Cell 1: Define generator pipeline
def read_data():
for i in range(1000000):
yield {'id': i, 'value': i * 2}
def filter_large(records):
for record in records:
if record['value'] > 1000:
yield record
def transform(records):
for record in records:
record['squared'] = record['value'] ** 2
yield record
# Cell 2: Execute pipeline and measure
import time
start = time.time()
pipeline = transform(filter_large(read_data()))
results = list(itertools.islice(pipeline, 100)) # Take first 100
print(f"Time: {time.time() - start:.4f}s")
print(f"Results: {len(results)}")
# Cell 3: Visualize with PyGWalker
import pygwalker as pyg
pyg.walk(results)FAQ
Conclusão
Os geradores em Python representam uma mudança fundamental da avaliação “ansiosa” (eager) para a avaliação preguiçosa (lazy), permitindo processamento eficiente em memória para conjuntos de dados que vão de milhares a bilhões de registros. Ao entender yield, expressões geradoras, o protocolo de iteradores e recursos avançados como send() e yield from, você consegue construir pipelines sofisticados de processamento de dados que escalam com facilidade.
Os principais pontos para lembrar:
- Geradores usam avaliação preguiçosa para minimizar o consumo de memória — frequentemente 99%+ de economia em relação a listas
- Use expressões geradoras para transformações simples e funções geradoras para lógica mais complexa
- Encadeie geradores para construir pipelines de processamento de dados eficientes em memória
- Aproveite
itertoolspara utilitários poderosos de iteração baseados em geradores - Escolha geradores para datasets grandes e iteração de passagem única; escolha listas para datasets pequenos que exigem acesso aleatório
Seja para processar logs massivos, fazer streaming de dados de API ou construir pipelines de ETL, geradores oferecem a performance e a eficiência de memória necessárias para processamento de dados em escala de produção. Domine esses padrões e você escreverá código Python que lida com conjuntos de dados de qualquer tamanho com elegância e eficiência.