Генераторы и yield в Python: миллионы строк без OOM

Обработка CSV на 8 миллионов строк через rows = list(reader) убивает процесс с MemoryError задолго до конца файла. Причина простая: список держит в памяти все объекты сразу, а Python на каждую строку тратит куда больше байт, чем она занимает на диске. Генераторы решают ровно эту проблему — они отдают элементы по одному и не хранят предыдущие. Ниже — как переписать конвейер обработки на yield, чем платишь за это и в каких случаях генератор не спасёт.

Примеры проверены на CPython 3.11 и 3.12. Синтаксис yield и модуль itertools работают одинаково начиная с Python 3.x, а генераторные выражения — с 2.4, так что версия здесь роли почти не играет; отдельно отмечу, где важна свежая.

Почему список съедает память, а генератор — нет

Когда вы вызываете функцию, которая возвращает список, она сначала строит его целиком и только потом отдаёт. Все объекты живут в памяти одновременно. Генератор устроен иначе: функция с yield при вызове не выполняется, а возвращает объект-генератор. Тело начинает работать, только когда вы запрашиваете следующий элемент через next() или цикл for. Дойдя до yield, функция замораживается — сохраняет своё состояние и отдаёт значение. Следующий запрос продолжает с того же места.

Практический смысл: в памяти в каждый момент находится одна строка, а не весь файл. Разница между обработкой файла в 100 МБ и в 100 ГБ по памяти — почти никакая.

Список отвечает на вопрос «дай мне всё сразу». Генератор отвечает «дай мне следующее, когда попросишь». Для потоковой обработки второй вопрос — единственно правильный.

Сравним лоб в лоб

# Плохо: весь файл в память
def read_all(path):
    with open(path, encoding="utf-8") as f:
        return f.readlines()   # список всех строк

# Хорошо: строки по одной
def read_lines(path):
    with open(path, encoding="utf-8") as f:
        for line in f:         # файловый объект сам итерируется лениво
            yield line.rstrip("\n")

Файловый объект в Python уже ленивый: цикл for line in f не читает весь файл. Обёртка с yield нужна, когда вы хотите добавить обработку — очистку, парсинг, фильтрацию — не ломая ленивость.

Строим конвейер из генераторов

Сила подхода — в цепочке. Каждый генератор берёт на вход другой генератор и отдаёт обработанные элементы дальше. Память не растёт, потому что через всю цепочку в момент времени проходит один элемент.

import csv

def read_rows(path):
    """Отдаёт строки CSV как словари, по одной."""
    with open(path, encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield row

def only_paid(rows):
    """Пропускает только оплаченные заказы."""
    for row in rows:
        if row["status"] == "paid":
            yield row

def with_total(rows):
    """Добавляет посчитанную сумму."""
    for row in rows:
        row["total"] = int(row["qty"]) * float(row["price"])
        yield row

# Ничего ещё не прочитано — только собран конвейер
pipeline = with_total(only_paid(read_rows("orders.csv")))

# Файл читается здесь, строка за строкой
total_sum = 0.0
for order in pipeline:
    total_sum += order["total"]

print(f"Сумма оплаченных заказов: {total_sum:.2f}")

Пока не начался финальный цикл for order in pipeline, файл не открыт и ни одна строка не прочитана. Это ключевое отличие от списковых включений: [with_total(...) for ...] материализовал бы всё сразу.

Генераторное выражение вместо функции

Для простых шагов не нужна отдельная функция — хватает генераторного выражения в круглых скобках:

# Список: держит все элементы
squares_list = [n * n for n in range(10_000_000)]   # ~400 МБ+

# Генератор: держит один
squares_gen = (n * n for n in range(10_000_000))     # байты

Разница только в скобках. Квадратные — список в памяти. Круглые — ленивый генератор.

itertools: инструменты для ленивых потоков

Стандартный модуль itertools даёт готовые генераторы, которые не ломают ленивость. Три самых полезных при обработке больших данных:

• islice — берёт срез из потока, не материализуя его. Замена list(gen)[10:20], которое прочитало бы всё.
• chain — склеивает несколько генераторов в один поток без объединения в список.
• batched — режет поток на кортежи фиксированного размера. Появился в Python 3.12; на 3.11 и раньше его нет, пишите свою функцию батчинга.

Батчинг нужен, когда вставлять в базу по одной строке слишком медленно, а грузить весь файл — нельзя. Компромисс — пачки по N строк:

from itertools import islice

def batched_compat(iterable, size):
    """Режет поток на пачки. Работает на Python 3.8+.
    На 3.12+ используйте itertools.batched."""
    it = iter(iterable)
    while True:
        chunk = list(islice(it, size))
        if not chunk:
            return
        yield chunk

# Вставляем в БД пачками по 1000, а не по одной
for batch in batched_compat(read_rows("orders.csv"), 1000):
    # cursor.executemany(...) на batch
    pass

Когда какой инструмент

Где подход ломается

Генератор — не бесплатная замена списку. Есть случаи, где он подведёт, и лучше знать о них до продакшена, а не во время инцидента.

• Генератор одноразовый. После полного прохода он исчерпан. Второй for по тому же объекту ничего не выдаст — без ошибки, просто пустой цикл. Это частый источник тихих багов: посчитали сумму, потом попытались посчитать количество по тому же генератору и получили ноль.
• Нет len() и индексации. len(gen) и gen[5] кинут TypeError. Если логике нужен размер набора или доступ по индексу — генератор не подходит.
• Исключение внутри рвёт всю цепочку. Ошибка парсинга на одной битой строке в середине файла остановит конвейер. Обрабатывайте try/except внутри генератора, иначе потеряете весь поток из-за одной строки.
• Ресурс закроется раньше времени. Если файл открыт внутри генератора через with, а вы прервали итерацию по break, файл закроется при сборке генератора — но момент вы не контролируете. Для критичных ресурсов закрывайте явно или используйте contextlib.closing.
• Отладка сложнее. Поставить точку останова и посмотреть «весь набор» нельзя — набора нет. Приходится либо материализовать кусок через list(islice(gen, 10)), либо логировать по ходу.

Тихий баг с повторным проходом

orders = read_rows("orders.csv")

total = sum(int(o["qty"]) for o in orders)   # прошли весь генератор
count = sum(1 for o in orders)               # генератор исчерпан -> count == 0

print(total, count)   # count будет 0, без единой ошибки

Если нужны и сумма, и количество — считайте их за один проход или материализуйте набор в список осознанно, приняв стоимость по памяти.

Что делать дальше

Перепишите самый прожорливый по памяти скрипт на генераторный конвейер и замерьте пиковое потребление до и после — например, через tracemalloc.get_traced_memory() из стандартной библиотеки или запустив процесс под наблюдением за RSS. На типичной потоковой обработке разница будет в порядки: с гигабайтов до десятков мегабайт.

Дальше смотрите в сторону asyncio и асинхронных генераторов (async def с yield), если узкое место не память, а ожидание сети или диска. А инвалидация и параллельная обработка потока — отдельная тема, здесь мы разбирали только память в один поток.