Импорт 200 тысяч строк в Django: почему save() в цикле убивает импорт

Импорт каталога поставщика в интернет-магазин на Django 5.x: файл на 200 тысяч строк, у каждой — товар, категория, остаток, цена. Первая версия скрипта работала так: читаем CSV построчно, для каждой строки Product.objects.get_or_create(). На локальной машине с SQLite прошло за пару минут на тестовом файле в 500 строк. На проде с PostgreSQL 16 и полным файлом скрипт шёл 38 минут и упирался в таймаут воркера. Разбираем, куда уходит время и как сократить его до трёх-четырёх минут без смены базы и железа.

Где именно теряется время

Наивный импорт делает одну и ту же ошибку многократно: на каждую строку файла отправляет отдельный запрос в базу, а то и несколько. При 200 тысячах строк это 200 тысяч сетевых раундтрипов до PostgreSQL, и каждый — это latency соединения плюс парсинг запроса плюс фиксация транзакции.

Посчитаем на пальцах. Если один раундтрип до базы стоит 1 мс (оптимистично для сети внутри одного дата-центра), то 200 тысяч запросов — это 200 секунд только на дорогу, без учёта работы самой базы. А get_or_create() — это два запроса: SELECT, потом при отсутствии INSERT. Уже 400 тысяч. Добавьте автокоммит после каждой операции — и PostgreSQL честно сбрасывает WAL на диск после каждой строки.

Проблема наивного импорта почти никогда не в скорости самой базы. Она в количестве обращений к ней. Сократите число раундтрипов на порядок — и получите ускорение на порядок, ничего больше не трогая.

Как замерить, а не гадать

Прежде чем оптимизировать, посмотрите, сколько запросов уходит. В Django 5.x для этого есть django.db.connection.queries (работает только при DEBUG=True) и reset_queries(). На проде DEBUG выключен, поэтому для боевого замера ставьте django-debug-toolbar на копии базы или логируйте на уровне PostgreSQL через log_min_duration_statement.

from django.db import connection, reset_queries
from django.test.utils import CaptureQueriesContext

# Считаем запросы на небольшом куске данных
with CaptureQueriesContext(connection) as ctx:
    import_chunk(rows[:1000])

print(f"Запросов на 1000 строк: {len(ctx.captured_queries)}")
# Если тут 2000 — значит по два запроса на строку, есть куда расти

Экстраполируйте: если на 1000 строк уходит 2000 запросов, на полном файле их будет 400 тысяч. Это и есть ваша цель для сокращения.

Переписываем: bulk_create вместо цикла save

Первое, что убирает большую часть раундтрипов — bulk_create(). Он собирает объекты в память и вставляет их одним (или несколькими) INSERT с множеством VALUES.

from itertools import islice
from myshop.models import Product

def chunked(iterable, size):
    """Режем итератор на куски по size элементов."""
    iterator = iter(iterable)
    while chunk := list(islice(iterator, size)):
        yield chunk

def import_products(rows):
    for chunk in chunked(rows, 1000):
        objects = [
            Product(
                sku=row["sku"],
                name=row["name"],
                price=row["price"],
                stock=row["stock"],
            )
            for row in chunk
        ]
        # batch_size ограничивает размер одного INSERT,
        # чтобы не упереться в лимит параметров запроса
        Product.objects.bulk_create(objects, batch_size=1000)

Здесь два уровня разбивки на куски намеренно. Внешний chunked контролирует, сколько объектов держим в памяти одновременно — иначе на 200 тысячах строк список Python-объектов съест несколько гигабайт. Внутренний batch_size контролирует, сколько строк уйдёт в один SQL-запрос.

Зачем нужен batch_size, если есть внешний chunk

PostgreSQL через драйвер psycopg ограничен числом параметров в одном запросе — исторически это 65 535 связанных параметров. Если у товара 8 полей, то один INSERT на 10 тысяч строк потребует 80 тысяч параметров и упадёт. batch_size разбивает вставку на безопасные порции. Держите его так, чтобы batch_size × число_полей оставалось заметно ниже лимита.

Проблема get_or_create: обновление существующих

Каталог импортируют повторно: цены и остатки меняются, часть товаров уже в базе. Здесь bulk_create в чистом виде не годится — он попытается вставить дубликаты и упрётся в уникальный индекс по sku.

В Django 5.x для этого есть upsert через параметры update_conflicts, unique_fields и update_fields в bulk_create(). Это транслируется в INSERT ... ON CONFLICT DO UPDATE для PostgreSQL.

Product.objects.bulk_create(
    objects,
    update_conflicts=True,
    unique_fields=["sku"],        # по чему определяем конфликт
    update_fields=["price", "stock", "name"],  # что обновляем при совпадении
    batch_size=1000,
)

Одна операция и вставляет новое, и обновляет существующее. Раньше для этого писали связку из bulk_create(ignore_conflicts=True) и отдельного bulk_update() — теперь достаточно одного вызова. Поведение update_conflicts появилось в Django 4.1, так что на 5.x оно точно есть; на более старых версиях сверьтесь с документацией своей версии.

Внешние ключи: категория товара

Если у каждого товара есть category = ForeignKey(Category), наивный код снова начнёт делать Category.objects.get_or_create(name=...) в цикле — и вы вернули раундтрипы, от которых избавились. Категорий обычно немного, поэтому загрузите их разом в словарь и обращайтесь к памяти, а не к базе.

# Один запрос: все категории в словарь name -> id
category_map = {
    c.name: c.id for c in Category.objects.all()
}

objects = []
for row in chunk:
    category_id = category_map.get(row["category"])
    if category_id is None:
        # новую категорию создаём отдельно и обновляем словарь
        category = Category.objects.create(name=row["category"])
        category_map[category.name] = category.id
        category_id = category.id
    objects.append(Product(
        sku=row["sku"],
        name=row["name"],
        category_id=category_id,  # присваиваем id напрямую, без объекта
    ))

Обратите внимание на category_id=... вместо category=.... Присваивание id напрямую не заставляет ORM тянуть объект категории из базы.

Транзакция и сигналы

Оберните весь импорт в одну транзакцию через transaction.atomic(). Без неё PostgreSQL коммитит после каждого запроса и постоянно сбрасывает WAL на диск. С одной транзакцией на весь импорт коммит происходит один раз в конце.

Отдельная ловушка: bulk_create() и bulk_update() не вызывают сигналы pre_save и post_save и не запускают метод save() модели. Если у вас в save() считается slug или в сигнале обновляется поисковый индекс — при массовом импорте это не сработает. Логику пересчёта придётся вызывать явно после импорта, отдельным пакетным проходом.

Что получилось: сравнение подходов

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

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

• Индексы во время вставки. Каждый INSERT обновляет все индексы таблицы. При загрузке в почти пустую таблицу иногда выгоднее удалить неуникальные индексы до импорта и пересоздать после. С данными в таблице это опасно и обычно не окупается.
• Триггеры на уровне базы. Если на таблице висят триггеры PostgreSQL, они отработают на каждую строку и bulk-вставка их не обойдёт.
• Валидация. bulk_create() не вызывает full_clean(). Мусорные данные из файла попадут в базу, если не проверить их до вставки в самом импортёре.
• Одна гигантская транзакция. Если импорт упадёт на 190-тысячной строке, откатится всё. Для очень больших файлов делите на транзакции покусочно и ведите лог, с какого места продолжать.

Что настроить после

Выносите импорт в management-команду, а не в view — на 200 тысячах строк любой HTTP-таймаут вас убьёт. Долгие импорты запускайте через Celery или как отдельный процесс, чтобы веб-воркеры не блокировались. После импорта отдельным проходом пересчитайте всё, что раньше делали сигналы: slug, денормализованные счётчики, поисковый индекс. И включите на копии базы log_min_duration_statement, чтобы увидеть, какие запросы реально самые медленные — иногда узкое место не там, где кажется.