Потоковая обработка данных: порции, генераторы и память

Прямой ответ

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

Базовый механизм в Python — генератор: функция с yield возвращает значения по одному и сохраняет состояние между вызовами.

def read_rows(path: str, batch_size: int = 5000):
    """Читает CSV порциями, не загружая файл целиком."""
    batch = []
    with open(path, newline="", encoding="utf-8") as fh:
        for line in fh:
            batch.append(parse(line))
            if len(batch) == batch_size:
                yield batch
                batch = []
    if batch:
        yield batch

Цепочка генераторов образует конвейер: каждая стадия обрабатывает элемент и передаёт дальше, не накапливая промежуточный результат.

def transform(rows):
    for row in rows:
        if row["status"] == "active":
            yield normalize(row)

def load(rows, chunk=1000):
    buffer = []
    for row in rows:
        buffer.append(row)
        if len(buffer) == chunk:
            insert_many(buffer)
            buffer.clear()
    if buffer:
        insert_many(buffer)

load(transform(read_rows("export.csv")))

Важно: конвейер ленивый. Ни одна стадия не начинает работу, пока потребитель не запросит следующий элемент. Это даёт естественный backpressure: если запись в базу медленная, чтение автоматически замедляется, и память не растёт. О механизме в статье про backpressure.

Ключевые приёмы:

from itertools import batched

with psycopg2.connect(dsn) as conn:
    with conn.cursor(name="big_cursor") as cur:      # серверный курсор
        cur.execute("SELECT id, payload FROM events WHERE created_at > %s", (since,))
        for batch in batched(cur, 10000):
            process(batch)

Практика: где теряется выигрыш

1. Случайная материализация. Вызов list(gen) или передача генератора в функцию, которая строит список, отменяет потоковость. Признак — рост памяти до размера всего набора.

rows = list(read_rows("export.csv"))   # файл снова загружен целиком
total = sum(r["amount"] for r in read_rows("export.csv"))   # потоково, память постоянна

2. Сортировка и дедупликация требуют всего объёма. Эти операции принципиально не потоковые: для сортировки нужны все данные. Если сортировка необходима, её выполняют на стороне источника (запрос с ORDER BY) или принимают ограничение.

3. Группировка по ключу с произвольным порядком. Если одинаковые ключи разбросаны по файлу, группировка требует хранения всех групп. Решение: предварительная сортировка источника, тогда достаточно держать одну группу.

4. Размер порции влияет на производительность. Мелкие порции дают больше накладных расходов на вызовы, крупные — больше памяти. Практический диапазон для записи в базу — от 500 до 10 000 строк; точное значение подбирают измерением.

5. Ошибки в середине потока. Если порция записана, а следующая упала, часть данных уже в базе. Нужен либо идемпотентный повтор, либо явная фиксация прогресса — о подходах в статье про ETL-пайплайны.

6. Транзакция на весь набор. Открытая транзакция на миллионы строк удерживает старые версии и блокирует очистку. Практика — фиксировать данные порциями.

Рекомендации:

1. Проверяйте память на большом объёме. Тест на 10 000 строк не покажет проблему, которая проявится на 100 млн.
2. Измеряйте долю времени на чтение и обработку. Если чтение доминирует, увеличение порции или параллельное чтение даст больший эффект, чем оптимизация преобразований.
3. Фиксируйте прогресс. При длительной обработке сохраняйте позицию, чтобы после сбоя не начинать заново.
4. Не смешивайте потоковость и кэширование всего. Мемоизация результатов по уникальному ключу накапливает данные; подробнее — в статье про мемоизацию.
5. Отделяйте извлечение от преобразования. Конвейер из независимых стадий легче тестировать: каждая стадия — функция из итератора в итератор.

Практический сценарий: импорт выгруженного файла исчерпал память

Контекст: ежемесячный импорт выгрузки от партнёра выполнялся через pandas.read_csv. На файле 2,1 ГБ процесс завершался с ошибкой выделения памяти; ранее файл был втрое меньше.

1. Диагностика. Загрузка строила DataFrame целиком: пиковое потребление составило около 14 ГБ из-за типов по умолчанию и копий при преобразовании.
2. Временное решение. Увеличение лимита памяти помогло, но рост файла продолжался, поэтому решение было непригодным.
3. Постоянное решение. Импорт переписали на потоковую схему: построчное чтение файла, порции по 5 000 строк, преобразование в генераторе, запись пачками через execute_values с фиксацией после каждой пачки.
4. Дополнительно. Добавили фиксацию последней обработанной строки в таблице состояния: после сбоя импорт продолжается с места остановки, а не начинается заново. Записи делают идемпотентными по внешнему ключу, чтобы повтор не создавал дубликаты.
5. Результат. Пиковое потребление памяти — 180 МБ независимо от размера файла; время импорта 2,1 ГБ составило 6 минут против прежних 11 минут с учётом подкачки.

Вывод: потоковая обработка убирает зависимость от объёма входных данных. Основной барьер — привычка загружать данные целиком, а не техническая сложность: переписывание заняло меньше дня.

FAQ

Когда потоковая обработка не нужна? Когда объём заведомо мал и помещается в память с запасом. Потоковый код сложнее читать и отлаживать; применять его стоит там, где объём растёт или изначально велик.

Чем генератор отличается от списка в плане памяти? Список хранит все элементы одновременно, генератор — только текущее состояние. Для набора из миллиона строк разница составляет гигабайты против килобайт.

Как обрабатывать данные, требующие доступа к предыдущим элементам? Через скользящее окно фиксированного размера, например очередь с ограничением. Если требуется доступ ко всей истории, потоковость невозможна без внешнего хранилища.

Можно ли распараллелить потоковый конвейер? Да, разделив источник на независимые части и обрабатывая их в процессах. При этом порядок элементов перестаёт гарантироваться, что нужно учитывать при записи и фиксации прогресса.

Источники: Python docs (itertools, generators), psycopg2 docs (server-side cursors) — ссылки в шапке.