Импорт больших данных в Django: чек-лист перед продом

Задача звучит просто: загрузить в базу пять миллионов строк из выгрузки поставщика. На тестовом датасете в 10 000 записей скрипт отработал за секунду, и его отправили в прод. Там он съел всю память процесса, упал на строке 3 200 000, а откатившаяся транзакция заблокировала таблицу на время отката. Импорт пришлось начинать заново — но данные уже частично поменяли состояние связанных сущностей.

Наивный for row in rows: Model.objects.create(...) плох не потому, что медленный, а потому что делает три вещи одновременно неправильно: держит весь набор в памяти, шлёт по одному INSERT на строку и оборачивает всё в неявные автокоммиты либо в одну гигантскую транзакцию. Ниже — чек-лист того, что проверить до запуска на боевом объёме. Ориентир — Django 5.x и PostgreSQL 16, но большинство пунктов версионно-нейтральны.

Память: откуда берётся утечка на ровном месте

Проблема номер один на больших импортах — не скорость, а память. Есть два типичных источника её роста.

Первый — чтение всего источника целиком. list(csv.reader(f)) или pandas.read_csv без chunksize поднимают весь файл в RAM. Читайте потоково.

Второй, менее очевидный — SQL-лог Django. При DEBUG = True каждый выполненный запрос копится в django.db.connection.queries. На миллионах INSERT это отдельная утечка на пустом месте.

Первое, что стоит проверить в упавшем по памяти импорте, — не запущен ли он с DEBUG=True. Список выполненных запросов растёт линейно с их количеством и на длинном скрипте съедает память быстрее самих данных.

Импорт всегда запускайте с DEBUG = False. Если по какой-то причине это невозможно, периодически сбрасывайте лог через reset_queries() из django.db.

Читаем источник итератором, а не списком

import csv

def read_rows(path):
    # генератор: в памяти держится одна строка, а не весь файл
    with open(path, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield row

bulk_create вместо цикла create

Каждый .create() — это отдельный round-trip к базе. Пять миллионов сетевых обменов даже по localhost стоят дороже самой вставки. bulk_create склеивает записи в один INSERT с множеством VALUES.

Ключевой параметр — batch_size. Без него Django на PostgreSQL попытается вставить весь список одним запросом, и вы упрётесь либо в память, либо в лимит параметров. Разумный диапазон — от 500 до нескольких тысяч строк на батч; точное значение зависит от количества полей и подбирается замером.

from itertools import islice
from myapp.models import Product

BATCH_SIZE = 2000

def import_products(path):
    rows = read_rows(path)
    while True:
        chunk = list(islice(rows, BATCH_SIZE))
        if not chunk:
            break
        objs = [
            Product(
                sku=r["sku"],
                title=r["title"],
                price=r["price"],
            )
            for r in chunk
        ]
        Product.objects.bulk_create(objs, batch_size=BATCH_SIZE)

Что bulk_create не делает, и это важно проверить заранее:

• не вызывает метод save() модели и не шлёт сигналы pre_save/post_save. Если на них навешана логика (пересчёт полей, обновление поиска) — она не отработает;
• по умолчанию не заполняет pk у объектов на всех бэкендах одинаково. На PostgreSQL первичные ключи возвращаются, но полагаться на это без проверки на своей версии не стоит;
• не проверяет уникальность на уровне Python — только база кинет ошибку при конфликте.

Обновление существующих строк: bulk_update и upsert

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

Для сценария «вставить новое, обновить существующее по уникальному ключу» в свежих версиях Django у bulk_create есть аргументы update_conflicts, unique_fields и update_fields. Поведение зависит от версии Django и от бэкенда базы — перед использованием сверьтесь с разделом про bulk_create в официальной документации вашей версии, потому что набор поддерживаемых аргументов менялся.

Транзакции: не заворачивайте весь импорт в один atomic

Соблазн обернуть весь скрипт в with transaction.atomic() понятен: либо всё, либо ничего. На миллионах строк это создаёт две проблемы. Транзакция удерживает блокировки и растит WAL; откат такой транзакции при падении на последнем батче может занять столько же, сколько сама вставка.

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

from django.db import transaction

def import_products(path):
    rows = read_rows(path)
    batch_num = 0
    while True:
        chunk = list(islice(rows, BATCH_SIZE))
        if not chunk:
            break
        batch_num += 1
        objs = [Product(sku=r["sku"], title=r["title"], price=r["price"]) for r in chunk]
        try:
            with transaction.atomic():
                Product.objects.bulk_create(objs, batch_size=BATCH_SIZE)
        except Exception as exc:
            # логируем номер батча, чтобы перезапуститься с него
            print(f"batch {batch_num} failed: {exc}")
            raise

Идемпотентность важнее скорости

Импорт упадёт — вопрос когда, а не если. Скрипт должен переживать повторный запуск без дублей. Обеспечьте это на уровне базы: уникальный индекс по бизнес-ключу (например, по sku) плюс стратегия upsert. Тогда повторный прогон не создаст вторые копии, а обновит существующие.

Индексы и триггеры на время загрузки

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

Это оправдано для одноразовой миграции в пустую или почти пустую таблицу и опасно на таблице под живым трафиком: без индекса читающие запросы деградируют. Уникальные индексы, отвечающие за идемпотентность, не трогайте.

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

Самый быстрый путь на PostgreSQL — команда COPY напрямую, мимо ORM. Она не знает про модели, сигналы и валидацию Django, зато на порядок обгоняет любой Python-цикл. Берите её, когда данные уже чистые и вся логика — просто «положить строки в таблицу».

Что мерить и как

Не оптимизируйте вслепую. Прогоните импорт на репрезентативном срезе (хотя бы 5–10% боевого объёма) и снимите три цифры: пиковое потребление памяти процессом, время на батч, число запросов к базе. Число запросов удобно ловить через CaptureQueriesContext из django.test.utils на небольшом фрагменте — если на один батч приходится больше запросов, чем вы ожидали, где-то прячется дополнительный SELECT (частый случай — обращение к ForeignKey внутри цикла подготовки объектов).

Внешние ключи заполняйте по *_id, а не по объекту: Product(category_id=42) вместо Product(category=some_category), который потянет за собой лишний запрос или заставит вас держать все категории в памяти.

Что сделать перед запуском на проде

1. Прогнать на срезе данных, снять память, время батча и число запросов.
2. Убедиться, что DEBUG = False.
3. Проверить идемпотентность: два прогона подряд не создают дублей.
4. Решить судьбу транзакций — атомарность на батч, не на весь импорт.
5. Договориться о времени: под живым трафиком снимать индексы нельзя.
6. Настроить логирование номера батча, чтобы перезапуститься с точки падения, а не с нуля.

Дальше — вынесите импорт в management-команду или в фоновую задачу (Celery, RQ), а не запускайте руками из shell. В проде смотрите на рост WAL и на блокировки во время загрузки: pg_stat_activity покажет, не встал ли импорт в очередь за чужой транзакцией.