Обработка потока данных в Python: чек-лист приёмов

Скрипт читал CSV на 300 МБ — работал. Тот же скрипт на файле в 12 ГБ съел всю оперативку и умер с MemoryError. Проблема не в объёме данных, а в том, что pd.read_csv(path) и data = file.read() тянут в память всё сразу. Поток данных — это когда вы обрабатываете элементы по одному или пачками, не держа весь набор в RAM. Ниже — чек-лист приёмов, которые ломаются реже всего, и разбор, где каждый из них подводит. Всё проверено на CPython 3.11+.

Базовое правило: генератор вместо списка

Список материализует все элементы сразу. Генератор отдаёт по одному. Разница — вся память против одного элемента за раз.

# Плохо: весь файл в списке строк
lines = open('big.log').readlines()  # 12 ГБ в RAM
errors = [line for line in lines if 'ERROR' in line]

# Хорошо: файловый объект — уже итератор строк
with open('big.log', encoding='utf-8') as f:
    errors = (line for line in f if 'ERROR' in line)
    for line in errors:  # читаем по строке
        process(line)

Ключевой момент: генераторное выражение в круглых скобках (x for x in ...) ленивое, а в квадратных [x for x in ...] — нет. Одна пара скобок решает, влезет ли задача в память.

Если вы где-то написали list(generator), чтобы «просто посмотреть, что там», — вы отменили всю экономию памяти. В проде такой вызов легко переживает ревью и убивает сервис на большом входе.

Чек-лист: что проверить в пайплайне на потоке

1. Вход читается лениво. Файл — итерируйте объект напрямую, не .read() и не .readlines(). Для CSV — csv.reader, а не загрузка целиком. Для JSON-массива на гигабайты — потоковый парсер (см. ниже), а не json.load.
2. Между этапами передаётся генератор, а не список. Каждый return [...] в середине конвейера обнуляет ленивость. Возвращайте генератор или используйте yield.
3. Ни одного скрытого list(), sorted(), len() на потоке. Все три требуют полной материализации. sorted обязан собрать всё, чтобы отсортировать, — сортировка потока целиком в памяти невозможна без внешней сортировки.
4. Итератор не проходится дважды. Генератор одноразовый: после первого прохода он пуст. Если данные нужны повторно — либо читайте источник заново, либо честно материализуйте (и признайте, что это уже не поток).
5. Ошибки не рвут весь поток. Одна битая строка в середине 50 ГБ не должна ронять обработку предыдущих 49. Оборачивайте обработку элемента в try/except внутри цикла.
6. Ресурсы закрываются. Файлы, соединения, сетевые сокеты — через with или contextlib.closing. Генератор, который открыл файл и не закрыл, оставит дескриптор висеть.
7. Есть обработка пачками, где это выгодно. Запись в БД по одной строке — это N сетевых round-trip. Пачками по 1000 — в тысячу раз меньше.

Пачки: между «по одному» и «всё сразу»

Чистый поток по одному элементу упирается в накладные расходы: вставка в БД построчно, HTTP-запрос на каждый элемент. Компромисс — батчинг. В Python 3.12 для этого появилась готовая функция itertools.batched.

# Python 3.12+
from itertools import batched

with open('users.csv', encoding='utf-8') as f:
    reader = csv.reader(f)
    next(reader)  # пропустить заголовок
    for chunk in batched(reader, 1000):
        # chunk — кортеж из максимум 1000 строк
        insert_many(chunk)  # одна вставка вместо 1000

Если вы на 3.11 или ниже, itertools.batched ещё нет — соберите пачки вручную:

# Работает на 3.8+
from itertools import islice

def batched(iterable, size):
    it = iter(iterable)
    while batch := tuple(islice(it, size)):
        yield batch

Размер пачки — компромисс между памятью и числом операций. 1000 строк по 200 байт — это 200 КБ в RAM за раз, приемлемо почти везде. Пачка в миллион строк снова вернёт вас к MemoryError.

Потоковый разбор форматов

С построчными форматами (логи, CSV, JSONL) поток тривиален: одна строка — один элемент. Проблема начинается с форматов, где документ монолитный.

• JSON-массив на гигабайты. json.load строит весь объект в памяти. Для потокового разбора нужна сторонняя библиотека ijson — она отдаёт элементы по мере чтения. Уточните актуальный синтаксис в её документации на PyPI: имена функций менялись между версиями.
• JSONL (JSON Lines). Один валидный JSON на строку. Идеальный формат для потока: json.loads(line) в цикле по файлу. Если формат вашего источника выбираете вы — выбирайте JSONL, а не гигантский массив.
• XML. xml.etree.ElementTree.iterparse из стандартной библиотеки читает потоком. Важная деталь: после обработки элемента вызывайте elem.clear(), иначе дерево копится в памяти и вся ленивость теряется.

Сравнение подходов к чтению

Что ломается на практике

Генератор молча проглатывает исключения

Если исключение возникает внутри генератора, оно всплывёт не в момент создания генератора, а в момент, когда цикл дойдёт до проблемного элемента. Трассировка укажет на for, а не на настоящую причину. Ставьте обработку внутри, ближе к источнику.

Двойной проход по одноразовому итератору

records = (parse(line) for line in open('data.txt'))
total = sum(1 for _ in records)   # прошли до конца
first = next(records, None)       # None — генератор уже пуст

Классическая ошибка: посчитали длину, потом попытались обработать — и получили пустоту. Если нужны и подсчёт, и обработка — считайте в том же цикле.

Незакрытый ресурс в цепочке генераторов

Функция вернула (line for line in open(path)) — файл открыт, но нигде не закрывается явно. При исключении в потребителе дескриптор повиснет. Оборачивайте открытие в with внутри генераторной функции с yield: тогда выход из блока закроет файл даже при обрыве потока.

Пачка последняя — неполная

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

Как убедиться, что поток действительно потоковый

Не полагайтесь на «вроде работает на тестовом файле». Проверьте память под нагрузкой:

• Запустите на входе, который заведомо больше RAM, и смотрите RSS процесса — он должен держаться примерно постоянным, а не расти линейно с прочитанным объёмом. Линейный рост означает, что где-то данные копятся.
• Для точечного профилирования памяти используйте tracemalloc из стандартной библиотеки: снимите снимок в начале и через N элементов, сравните.
• Проверьте поведение на битой строке в середине входа: пайплайн должен либо пропустить её и продолжить, либо упасть с понятной ошибкой, но не молча потерять хвост.

Следующий шаг после того, как поток держит память: параллелизм. Обработка независимых элементов раскладывается на процессы через concurrent.futures.ProcessPoolExecutor — но это уже другая тема, здесь мы разбирали только память и целостность одного потока.