Миграция данных в Python: сравнение четырёх подходов

Схему в проекте на Django или Alembic вы двигаете аккуратно: makemigrations, ревью, откат при желании. А вот когда нужно переложить два миллиона строк из старой таблицы в новую, разбить одно поле на два или дозаполнить колонку по внешнему источнику — начинается самодеятельность. Кто-то пишет UPDATE прямо в консоли базы, кто-то гоняет разовый скрипт с ноутбука по продовому соединению. И то и другое рано или поздно роняет базу под нагрузкой или оставляет данные в половинчатом состоянии.

Разберём четыре подхода к миграции данных в Python-проекте и цену каждого: транзакционность, блокировки, возможность откатить и остановить. Речь именно про перелив и трансформацию данных, не про изменение структуры таблиц — DDL-миграции это отдельная тема.

Задача и почему очевидное решение бьёт по проду

Допустим, было поле full_name, стало два: first_name и last_name. Колонки вы добавили обычной миграцией. Осталось разложить существующие значения. Очевидное решение — один запрос на всю таблицу:

# так делать не надо на большой таблице
Users.objects.update(...)  # один UPDATE на все строки

На таблице в пару тысяч строк это незаметно. На миллионах строк один UPDATE в PostgreSQL держит долгую транзакцию, пишет весь объём в WAL, а в MySQL с InnoDB может забрать блокировки на диапазоны и заставить остальные запросы ждать. Пользователи в этот момент получают таймауты. Отменить процесс на середине нельзя без отката всей транзакции, а откат миллиона строк — это ещё столько же работы.

Отсюда четыре осмысленных подхода, отличающихся тем, где живёт логика и как вы дробите работу.

Подход 1. Data migration внутри фреймворка

В Django это RunPython внутри файла миграции, в Alembic — код в функции upgrade(). Логика едет вместе с кодом, попадает в тот же граф зависимостей, что и структурные миграции, и применяется одной командой на деплое.

Django 5.x, схематичный фрагмент внутри миграции:

from django.db import migrations


def split_full_name(apps, schema_editor):
    # берём историческую модель, а не импортируем актуальную
    User = apps.get_model("accounts", "User")
    batch_size = 2000
    qs = User.objects.exclude(full_name="").only("id", "full_name")
    to_update = []
    for user in qs.iterator(chunk_size=batch_size):
        parts = user.full_name.split(" ", 1)
        user.first_name = parts[0]
        user.last_name = parts[1] if len(parts) > 1 else ""
        to_update.append(user)
        if len(to_update) >= batch_size:
            User.objects.bulk_update(to_update, ["first_name", "last_name"])
            to_update.clear()
    if to_update:
        User.objects.bulk_update(to_update, ["first_name", "last_name"])


class Migration(migrations.Migration):
    dependencies = [("accounts", "0012_add_name_fields")]
    operations = [
        migrations.RunPython(split_full_name, migrations.RunPython.noop),
    ]

Ключевая деталь — apps.get_model вместо прямого импорта модели. Модель из models.py отражает состояние кода сейчас, а миграция должна работать с тем состоянием схемы, которое было на её момент. Импортируете актуальную модель — и через полгода, когда в ней исчезнет поле, миграция перестанет накатываться на чистой базе.

Цена. По умолчанию Django оборачивает миграцию в транзакцию (кроме баз без транзакционного DDL). Значит, весь перелив — одна транзакция, и все проблемы долгой транзакции из предыдущего раздела возвращаются. Дробление на bulk_update внутри не спасает от общего коммита в конце. Для больших таблиц транзакцию с миграции снимают через atomic = False у класса миграции, и тогда каждый батч нужно коммитить самому.

Подход 2. Management-команда или отдельный скрипт

Логику перелива выносят в management-команду Django (BaseCommand) или в самостоятельный скрипт, который запускают руками или из планировщика после деплоя. Миграция при этом только меняет структуру, а данные переливаются отдельным шагом.

Почему это часто лучше RunPython: вы полностью управляете транзакциями, можете делать паузы между батчами, логировать прогресс и остановить процесс по Ctrl+C без отката всего сделанного. Каждый батч коммитится сам.

from django.core.management.base import BaseCommand
from django.db import transaction
from accounts.models import User


class Command(BaseCommand):
    help = "Переливает full_name в first_name/last_name батчами"

    def handle(self, *args, **options):
        batch_size = 2000
        last_id = 0
        while True:
            # keyset-пагинация по id, без OFFSET
            batch = list(
                User.objects.filter(id__gt=last_id)
                .exclude(full_name="")
                .order_by("id")[:batch_size]
            )
            if not batch:
                break
            with transaction.atomic():
                for user in batch:
                    parts = user.full_name.split(" ", 1)
                    user.first_name = parts[0]
                    user.last_name = parts[1] if len(parts) > 1 else ""
                User.objects.bulk_update(
                    batch, ["first_name", "last_name"]
                )
            last_id = batch[-1].id
            self.stdout.write(f"Обработано до id={last_id}")

Пагинация здесь по id__gt, а не через OFFSET. OFFSET на глубоких страницах заставляет базу пролистывать и отбрасывать всё, что до смещения, и к концу таблицы каждый батч тормозит сильнее предыдущего. Keyset-пагинация по индексированному id держит скорость ровной.

Цена. Команда не привязана к графу миграций. Если забудете запустить её после деплоя, схема уедет вперёд, а данные останутся старыми. Ответственность за запуск и идемпотентность — на вас: повторный прогон должен не портить уже перелитые строки.

Подход 3. Чистый SQL батчами

Когда трансформация выражается в SQL без бизнес-логики на Python, гонять миллионы строк через ORM — лишние объекты и лишняя память. Тот же перелив можно сделать запросом, а дробление на батчи оставить снаружи.

from django.db import connection

def run_batch(last_id, batch_size=5000):
    with connection.cursor() as cur:
        cur.execute(
            """
            UPDATE users
            SET first_name = split_part(full_name, ' ', 1),
                last_name  = substr(full_name,
                             strpos(full_name, ' ') + 1)
            WHERE id IN (
                SELECT id FROM users
                WHERE id > %s AND full_name <> ''
                ORDER BY id LIMIT %s
            )
            RETURNING id
            """,
            [last_id, batch_size],
        )
        rows = cur.fetchall()
    return rows[-1][0] if rows else None

Функции split_part, substr, strpos — из PostgreSQL 16, в другой СУБД имена и поведение отличаются, проверяйте по документации своей базы. RETURNING есть в PostgreSQL; в MySQL его нет, придётся считать сдвиг иначе.

Цена. Максимальная скорость и минимум памяти, но вы теряете валидацию и сигналы ORM, а логику дублируете на SQL. Если правило разбора имени сложнее, чем разрез по первому пробелу, SQL быстро становится нечитаемым.

Подход 4. Фоновая задача через очередь

Если перелив длинный и не должен блокировать деплой, его отдают в Celery или другой воркер: миграция ставит одну задачу, а та сама разбивает работу на подзадачи-батчи и складывает их в очередь. Деплой не ждёт окончания, прогресс виден в мониторинге очереди.

Цена. Самая сложная в эксплуатации схема. Нужен работающий брокер, обработка повторных запусков задачи (at-least-once доставка означает, что батч может выполниться дважды), отслеживание, что все подзадачи завершились. Оправдана на действительно больших объёмах или когда перелив идёт в фоне на живом трафике.

Сравнение

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

Когда что выбирать

• RunPython — небольшие и средние таблицы, где важно, чтобы данные и схема доехали одной командой и откатились вместе. Порог по размеру зависит от базы и нагрузки, ориентируйтесь на длительность транзакции, а не на число строк.
• Команда или скрипт — большие таблицы, где нужна пауза, прогресс и остановка без отката. Дефолтный выбор для перелива, который идёт минуты и дольше.
• SQL батчами — трансформация выражается запросом без Python-логики, объёмы большие, память в приоритете.
• Очередь — перелив длится часами, идёт на живом трафике и не должен задерживать релиз.

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

• Неидемпотентный перелив. Скрипт упал на середине, вы запустили заново — и половина строк обработалась дважды. Разбиение по первому пробелу это переживёт, а вот balance = balance + bonus при повторе начислит бонус второй раз. Делайте операции идемпотентными или отмечайте обработанные строки флагом.
• Импорт актуальной модели в RunPython. Работает сегодня, ломает накат на чистой базе завтра. Только apps.get_model.
• OFFSET-пагинация. Первые батчи быстрые, последние ползут. Keyset по индексированному ключу.
• Долгая транзакция под нагрузкой. В PostgreSQL раздувает WAL и мешает автовакууму, в MySQL держит блокировки. Коммитьте каждый батч.
• Забытый запуск команды. Схема ушла вперёд, данные старые. Ставьте проверку в деплой-пайплайн или health-check.

Что настроить дальше

Перед запуском на проде прогоните перелив на копии боевой базы и замерьте время одного батча и его влияние на конкурентные запросы. Подберите batch_size так, чтобы одна транзакция закрывалась быстрее вашего порога по блокировкам, — универсального числа нет, для широких строк 2000 может оказаться много, для узких 5000 мало. В мониторинге на время перелива следите за длительностью самой старой транзакции и за очередью блокировок в базе. И заведите способ поставить перелив на паузу — рано или поздно он понадобится в неудобный момент.