Как синхронизировать базы данных при переносе данных из одной базы в другую?

Ответ

Синхронизация баз данных при переносе данных из одной в другую — это сложная задача, требующая внимания к целостности данных, производительности и минимизации простоя. В Go это обычно реализуется путем комбинации различных подходов:

Пошаговый перенос с контролем ошибок и идемпотентностью: Прямые распределенные транзакции между разными базами данных крайне сложны и редко используются. Вместо этого, фокус делается на надежном переносе данных с возможностью повторного выполнения операций без дублирования (идемпотентность).

package main

import (
    "database/sql"
    "fmt"
    "log"
    _ "github.com/go-sql-driver/mysql" // Пример драйвера
)

// Предполагаем, что sourceDB и targetDB уже инициализированы
func transferData(sourceDB, targetDB *sql.DB) error {
    // Начнем транзакцию в исходной базе данных для чтения
    // Это гарантирует, что мы читаем согласованный набор данных из sourceDB
    txSource, err := sourceDB.Begin()
    if err != nil {
        return fmt.Errorf("не удалось начать транзакцию в sourceDB: %w", err)
    }
    defer txSource.Rollback() // Откат в случае ошибки или до успешного коммита

    rows, err := txSource.Query("SELECT id, name, value FROM source_table ORDER BY id")
    if err != nil {
        return fmt.Errorf("ошибка запроса к source_table: %w", err)
    }
    defer rows.Close()

    // Начнем транзакцию в целевой базе данных для записи
    // Это гарантирует атомарность записи батча данных в targetDB
    txTarget, err := targetDB.Begin()
    if err != nil {
        return fmt.Errorf("не удалось начать транзакцию в targetDB: %w", err)
    }
    defer txTarget.Rollback() // Откат в случае ошибки или до успешного коммита

    stmt, err := txTarget.Prepare("INSERT INTO target_table (id, name, value) VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE name=VALUES(name), value=VALUES(value)")
    if err != nil {
        return fmt.Errorf("ошибка подготовки запроса в targetDB: %w", err)
    }
    defer stmt.Close()

    for rows.Next() {
        var id int
        var name, value string
        if err := rows.Scan(&id, &name, &value); err != nil {
            return fmt.Errorf("ошибка сканирования строки из sourceDB: %w", err)
        }

        // Выполняем вставку/обновление в целевой БД
        // Использование ON DUPLICATE KEY UPDATE (для MySQL) или UPSERT (для PostgreSQL) обеспечивает идемпотентность
        _, err = stmt.Exec(id, name, value)
        if err != nil {
            return fmt.Errorf("ошибка вставки/обновления в targetDB: %w", err)
        }
    }

    if err := rows.Err(); err != nil {
        return fmt.Errorf("ошибка итерации по строкам sourceDB: %w", err)
    }

    // Коммитим транзакцию в целевой БД
    if err := txTarget.Commit(); err != nil {
        return fmt.Errorf("не удалось закоммитить транзакцию в targetDB: %w", err)
    }

    // Коммитим транзакцию в исходной БД (если это необходимо для логирования или изменения состояния)
    // В данном случае, если мы только читаем, коммит не обязателен, но Rollback() в defer все равно сработает.
    // Если бы мы помечали данные как перенесенные, то коммит был бы важен.
    if err := txSource.Commit(); err != nil {
        return fmt.Errorf("не удалось закоммитить транзакцию в sourceDB: %w", err)
    }

    return nil
}

func main() {
    // Пример инициализации (замените на свои данные)
    sourceDB, err := sql.Open("mysql", "user:password@tcp(127.0.0.1:3306)/source_db")
    if err != nil {
        log.Fatalf("Не удалось подключиться к sourceDB: %v", err)
    }
    defer sourceDB.Close()

    targetDB, err := sql.Open("mysql", "user:password@tcp(127.0.0.1:3306)/target_db")
    if err != nil {
        log.Fatalf("Не удалось подключиться к targetDB: %v", err)
    }
    defer targetDB.Close()

    if err := transferData(sourceDB, targetDB); err != nil {
        log.Fatalf("Ошибка переноса данных: %v", err)
    }
    fmt.Println("Данные успешно перенесены.")
}

Идемпотентность: Используйте INSERT ... ON DUPLICATE KEY UPDATE (MySQL) или INSERT ... ON CONFLICT (target) DO UPDATE (PostgreSQL) для обеспечения того, что повторный запуск операции не приведет к дублированию данных, а обновит существующие.
Транзакции: Используйте транзакции внутри каждой базы данных для обеспечения атомарности операций чтения из источника и записи в цель. Это не распределенная транзакция, но повышает надежность.

Batch-обработка (пакетирование): Разбивайте данные на небольшие пачки (например, по 1000-10000 строк) для уменьшения нагрузки на память и сеть, а также для более эффективной обработки ошибок. Каждая пачка может быть обработана в отдельной транзакции в целевой БД.

const batchSize = 1000
offset := 0

for {
    // Чтение батча из sourceDB
    rows, err := sourceDB.Query("SELECT * FROM table LIMIT ? OFFSET ?", batchSize, offset)
    if err != nil {
        return err
    }
    // Проверка, есть ли еще строки
    if !rows.Next() {
        rows.Close()
        break // Нет больше данных
    }
    // Обработка батча (например, сбор данных в слайс и затем массовая вставка в targetDB)
    // ...
    rows.Close()
    offset += batchSize
}

Использование специализированных инструментов и подходов:
- SQL-дампы: Для больших объемов данных или полного переноса можно использовать утилиты, такие как pg_dump/pg_restore (PostgreSQL) или mysqldump/mysql (MySQL). Они создают SQL-скрипты, которые затем выполняются в целевой БД.
- ETL-решения (Extract, Transform, Load): Для сложных сценариев с трансформацией данных, очисткой и оркестрацией существуют специализированные ETL-инструменты (например, Apache NiFi, Talend, Airflow).
- Change Data Capture (CDC): Для непрерывной синхронизации в реальном времени можно использовать CDC-инструменты (например, Debezium, Maxwell), которые отслеживают изменения в логах транзакций исходной БД и реплицируют их в целевую.
- Репликация на уровне БД: Некоторые СУБД поддерживают встроенные механизмы репликации (логическая/физическая репликация), которые могут быть использованы для поддержания синхронизации между экземплярами.

Стратегии минимизации простоя (для миграции):

Синхронизация исторических данных: Сначала перенесите основную часть данных, которые не меняются часто.
Перенос дельты: Затем перенесите изменения, произошедшие за время первого переноса.
Блокировка записи (короткая): На короткое время заблокируйте запись в исходную БД, перенесите последнюю дельту изменений, убедитесь в согласованности.
Переключение приложения: Переключите приложение на работу с новой (целевой) БД.
Двунаправленная запись (Dual Write): В некоторых случаях, для минимизации простоя, можно временно настроить приложение на запись в обе базы данных одновременно, пока идет миграция, а затем переключить чтение на новую БД.