В чем разница между миграцией и миграцией данных?

Разница между миграцией и миграцией данных

Миграция (Schema Migration)

Миграция — это изменение структуры базы данных (schema). Это процесс версионирования изменений в таблицах, индексах, constraints и других структурных элементах БД.

Примеры миграций:

-- Добавление нового столбца
ALTER TABLE users ADD COLUMN phone_number VARCHAR(20);

-- Удаление столбца
ALTER TABLE users DROP COLUMN legacy_field;

-- Изменение типа данных
ALTER TABLE orders MODIFY COLUMN total DECIMAL(12, 2);

-- Добавление индекса
CREATE INDEX idx_users_email ON users(email);

-- Создание новой таблицы
CREATE TABLE notifications (
    id UUID PRIMARY KEY,
    user_id UUID NOT NULL REFERENCES users(id),
    message TEXT,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Добавление constraints
ALTER TABLE users ADD CONSTRAINT unique_email UNIQUE(email);

В проекте Prepbro миграции хранятся в migrations/ и используют Goose:

-- 0001_initial_schema.sql
CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    email VARCHAR(255) NOT NULL UNIQUE,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- 0002_add_phone.sql
ALTER TABLE users ADD COLUMN phone_number VARCHAR(20);

Миграции версионируются и воспроизводимы:

# Накатить все миграции
goose up

# Откатить одну миграцию
goose down

# Статус миграций
goose status

Миграция данных (Data Migration)

Миграция данных — это преобразование существующих данных в новый формат. Это может быть:

• Заполнение нового столбца значениями
• Трансформация данных
• Переход с одной структуры на другую
• Очистка данных
• Денормализация или нормализация

Примеры миграций данных:

# Пример 1: Заполнить новый столбец на основе старого
UPDATE users SET phone_number = '+7' || phone WHERE phone IS NOT NULL;

# Пример 2: Заполнить значение по умолчанию
UPDATE products SET status = 'active' WHERE status IS NULL;

# Пример 3: Разделить данные
UPDATE users SET first_name = SPLIT_PART(full_name, ' ', 1);
UPDATE users SET last_name = SPLIT_PART(full_name, ' ', 2);

Сценарий: добавление нового поля

Шаг 1: Миграция (изменение схемы)

-- migrations/0003_add_user_phone.sql
ALTER TABLE users ADD COLUMN phone_number VARCHAR(20) DEFAULT NULL;
ALTER TABLE users ADD CONSTRAINT valid_phone CHECK (phone_number IS NULL OR phone_number ~ '^\+?[0-9\s\-\(\)]+$');

Шаг 2: Миграция данных (заполнение данных)

# scripts/migrate_phone_data.py
import psycopg2
from datetime import datetime

conn = psycopg2.connect(
    host='localhost',
    database='prepbro',
    user='postgres',
    password='password'
)

cursor = conn.cursor()

# Вариант 1: Просто SQL запрос
cursor.execute("""
    UPDATE users 
    SET phone_number = '+7' || old_phone 
    WHERE old_phone IS NOT NULL 
    AND phone_number IS NULL
""")

# Вариант 2: Трансформация с логикой
cursor.execute('SELECT id, old_phone FROM users WHERE old_phone IS NOT NULL')
rows = cursor.fetchall()

for user_id, old_phone in rows:
    # Логика трансформации
    formatted_phone = format_phone_number(old_phone)
    
    cursor.execute(
        'UPDATE users SET phone_number = %s WHERE id = %s',
        (formatted_phone, user_id)
    )

conn.commit()
cursor.close()
conn.close()

Сценарий: разделение таблицы

Шаг 1: Миграция (создание новых таблиц)

-- migrations/0004_split_user_info.sql
CREATE TABLE user_profiles (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE UNIQUE,
    first_name VARCHAR(100),
    last_name VARCHAR(100),
    bio TEXT,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

Шаг 2: Миграция данных (переход данных)

# scripts/migrate_user_profiles.py
import psycopg2
from uuid import uuid4

conn = psycopg2.connect(...)
cursor = conn.cursor()

# Вариант 1: Одним SQL запросом
cursor.execute("""
    INSERT INTO user_profiles (user_id, first_name, last_name, bio)
    SELECT id, first_name, last_name, bio FROM users
    WHERE first_name IS NOT NULL OR last_name IS NOT NULL
""")

conn.commit()

# Проверка: сколько записей перенеслось?
cursor.execute('SELECT COUNT(*) FROM user_profiles')
count = cursor.fetchone()[0]
print(f'Перенесено {count} профилей')

cursor.close()
conn.close()

Ключевые различия

Паттерн: миграция + миграция данных

# 1. Добавили столбец в миграции
goose up

# 2. Заполнили данные скриптом
python scripts/migrate_phone_data.py

# 3. Проверили данные
psql -c 'SELECT COUNT(*) FROM users WHERE phone_number IS NOT NULL'

# 4. Теперь приложение может использовать новое поле

Безопасность при миграциях данных

# ✅ Хорошо: с проверками и откатом
def migrate_phone_data():
    conn = get_connection()
    cursor = conn.cursor()
    
    try:
        # Считаем сколько записей затронется
        cursor.execute('SELECT COUNT(*) FROM users WHERE old_phone IS NOT NULL')
        count = cursor.fetchone()[0]
        print(f'Будет затронуто {count} записей')
        
        # Проверка перед миграцией
        if count == 0:
            print('Нет данных для миграции')
            return
        
        # Миграция
        cursor.execute("""
            UPDATE users 
            SET phone_number = '+7' || old_phone 
            WHERE old_phone IS NOT NULL
        """)
        
        # Проверка после
        cursor.execute('SELECT COUNT(*) FROM users WHERE phone_number IS NOT NULL')
        migrated = cursor.fetchone()[0]
        assert migrated == count, f'Ошибка: мигрировано {migrated}, ожидалось {count}'
        
        conn.commit()
        print(f'Успешно мигрировано {migrated} записей')
        
    except Exception as e:
        conn.rollback()
        print(f'Ошибка: {e}')
        raise
    finally:
        cursor.close()
        conn.close()

Минимизация downtime при больших миграциях

# ✅ Миграция данных по частям (chunked migration)
def migrate_large_dataset():
    conn = get_connection()
    cursor = conn.cursor()
    
    CHUNK_SIZE = 10000
    offset = 0
    
    while True:
        cursor.execute(f"""
            SELECT id FROM users 
            WHERE phone_number IS NULL AND old_phone IS NOT NULL
            LIMIT {CHUNK_SIZE} OFFSET {offset}
        """)
        
        user_ids = [row[0] for row in cursor.fetchall()]
        if not user_ids:
            break
        
        # Обновляем маленькую порцию (меньше блокировок)
        cursor.execute("""
            UPDATE users 
            SET phone_number = '+7' || old_phone 
            WHERE id IN (" + ",".join([f"'{uid}'" for uid in user_ids]) + ")
        """)
        conn.commit()
        
        offset += CHUNK_SIZE
        print(f'Мигрировано {offset} записей')

Вывод

• Миграция = изменение структуры БД (schema migration)
• Миграция данных = преобразование существующих данных (data migration)
• Обычно идут вместе: сначала структура, потом данные
• Миграции версионируются в Goose, миграции данных — в Python скриптах
• Большие миграции данных нужно выполнять по частям для минимизации downtime