Пакетная обработка в Python: рабочие приёмы для больших объёмов

Задача звучит невинно: пройти по трём миллионам записей, обогатить каждую данными из внешнего API и записать в другую таблицу. Наивная реализация — цикл for row in rows с одиночным запросом на каждой итерации — на таких объёмах выполняется часами и падает по памяти, если вы предварительно загрузили всё в список. Пакетная обработка (batching) решает обе проблемы: вы работаете порциями фиксированного размера, контролируя и время, и потребление памяти. Ниже — рабочие приёмы для Python 3.11+ и типичные грабли.

Почему одиночная обработка проваливается

Проблем на самом деле две, и их часто путают.

• Память. rows = cursor.fetchall() или list(queryset) тянет весь результат в RAM. Миллион строк по килобайту — уже гигабайт, а объекты Python занимают заметно больше сырых данных.
• Накладные расходы на операцию. Каждый одиночный INSERT, HTTP-запрос или вызов внешней функции несёт фиксированную стоимость: round-trip до базы, установку соединения, сериализацию. Умножьте на миллион — и стоимость обвязки превысит стоимость полезной работы в десятки раз.

Batching бьёт по обеим: порции ограничивают память, а групповые операции (bulk insert, batch API) размазывают фиксированную стоимость на всю пачку.

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

Приём 1. Нарезка итерируемого на пачки

С Python 3.12 в стандартной библиотеке появился itertools.batched — он делает ровно то, что нужно, и не держит весь источник в памяти, если источник ленивый.

from itertools import batched

# Python 3.12+
# batched возвращает кортежи по batch_size элементов,
# последний кортеж может быть короче.
for chunk in batched(range(1_000_000), 5_000):
    process(chunk)  # chunk — кортеж из максимум 5000 элементов

Если вы на 3.11 или ниже, batched недоступен — напишите генератор сами. Это несколько строк и работает с любым итерируемым:

from itertools import islice

def batched(iterable, batch_size):
    """Аналог itertools.batched для Python < 3.12."""
    iterator = iter(iterable)
    while True:
        chunk = tuple(islice(iterator, batch_size))
        if not chunk:
            return
        yield chunk

Ключевой момент: islice берёт из итератора ровно batch_size элементов и не материализует остаток. Если вы передадите сюда список — весь список уже в памяти, и выигрыша по RAM не будет. Batching по памяти работает только когда источник ленивый: генератор, курсор с серверным стороны, построчное чтение файла.

Приём 2. Не грузить весь результат из базы

Самая частая ошибка — нарезать на пачки то, что уже целиком в памяти. Читать нужно порциями с самого начала.

psycopg (PostgreSQL)

Для psycopg 3.x серверный курсор объявляется через именованный курсор, тогда база отдаёт строки порциями, а не сваливает весь результат сразу.

import psycopg  # psycopg 3.x

with psycopg.connect("postgresql://user:YOUR_PASSWORD@localhost/db") as conn:
    # Серверный курсор: строки не грузятся все сразу
    with conn.cursor(name="big_read") as cur:
        cur.itersize = 5_000  # сколько строк тянуть за один round-trip
        cur.execute("SELECT id, payload FROM events")
        for chunk in batched(cur, 5_000):
            process(chunk)

Django ORM

В Django для чтения больших выборок есть QuerySet.iterator(chunk_size=...) — он не кеширует весь результат в QuerySet и на PostgreSQL использует серверный курсор.

from myapp.models import Event

# Django 4.x / 5.x
for chunk in batched(Event.objects.iterator(chunk_size=2_000), 2_000):
    process(chunk)

Проверьте актуальное поведение iterator() и параметра chunk_size в документации вашей версии Django — детали по серверным курсорам зависят от бэкенда БД.

Приём 3. Групповая запись вместо цикла INSERT

Чтение порциями бессмысленно, если вы потом пишете по одной строке. На запись работают те же принципы.

В Django для вставки используйте bulk_create, для обновления — bulk_update, оба принимают batch_size:

# Django 4.x / 5.x
objs = [Event(payload=x) for x in prepared_data]
Event.objects.bulk_create(objs, batch_size=1_000)

Помните про ограничения: bulk_create не вызывает метод save() модели и сигналы pre_save/post_save. Если на них завязана логика — batching её обойдёт молча, и это самый неприятный класс багов в такой миграции.

Приём 4. Одна транзакция на пачку

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

from django.db import transaction

for chunk in batched(source_iterator, 1_000):
    with transaction.atomic():
        Event.objects.bulk_create(
            [Event(payload=x) for x in chunk],
            batch_size=1_000,
        )
    # После выхода из atomic пачка зафиксирована на диске

Так при сбое вы теряете максимум одну незавершённую пачку. Добавьте логирование ID последней успешной пачки — тогда перезапуск продолжит с места обрыва, а не с начала.

Сравнение подходов

Как выбрать размер пачки

Единственно верного числа нет, но есть ориентиры:

• Начните с 1000–5000 и померяйте память процесса и время на пачку.
• Если строки тяжёлые (BLOB, JSON на десятки килобайт) — уменьшайте, память съедается быстрее.
• Для внешних API размер диктует лимит самого API на число элементов в одном запросе, а не ваши предпочтения.
• Слишком маленькая пачка (10–50) возвращает вас к проблеме накладных расходов.

Где batching ломается

• Источник не ленивый. Нарезали список — сэкономили только на групповых операциях, но не на памяти. Проверьте, что источник — генератор или курсор.
• Побочные эффекты внутри пачки. Если обработка одного элемента зависит от результата предыдущего, наивный batching даст гонку. Batching предполагает независимость элементов внутри пачки.
• Частичный успех пачки. Групповая вставка либо проходит целиком, либо падает целиком. Одна битая запись роняет всю пачку — нужна либо предварительная валидация, либо fallback на поштучную обработку упавшей пачки.
• Порядок и дубликаты при перезапуске. Если вы перезапускаете с последней пачки, убедитесь, что запись идемпотентна (ON CONFLICT в PostgreSQL или update_conflicts в Django), иначе получите дубли.

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

Соберите минимальный конвейер: ленивое чтение → batched → групповая запись в транзакции на пачку. Замерьте RSS процесса (psutil.Process().memory_info().rss) и время на пачку до и после — без цифр вы не поймёте, где узкое место: в чтении, обработке или записи. Если обработка внутри пачки CPU-bound и элементы независимы — следующий шаг это concurrent.futures.ProcessPoolExecutor на уровне пачек, но параллелизм — отдельная тема, здесь мы её не разворачиваем.