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

А, синхронизация баз, говоришь? Ну это, конечно, пиздец задача, если делать в лоб. Представь: две базы, как два упрямых осла, стоят в разных концах поляны, а тебе надо, чтобы они хуяк — и думали одинаково. И чтобы в процессе никто не обосрался с данными.

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

Вот, смотри на этот кусок кода, я его не трогал, он правильный:

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("Данные успешно перенесены.")
}

Видишь этот ON DUPLICATE KEY UPDATE? Это, блядь, волшебные слова! Если запись уже есть — он её обновит, а не вставит вторую. Вот и вся идемпотентность, в рот меня чих-пых! И транзакции в каждой базе свои — чтобы если что-то пошло не так, можно было откатиться, а не сидеть с разорванной жопой.

А теперь, если данных овердохуища, нельзя их все разом тащить. Представь, ты везешь мешок картошки — порвётся. Надо батчами, пачками! По 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-дампы — старый добрый mysqldump. Выгрузил всё в файлик, загрузил в новую базу. Просто, как три копейки, но для живых систем не всегда годится.
• ETL-системы (Extract, Transform, Load) — это уже серьёзно, типа Apache NiFi или Airflow. Там можно выстраивать целые конвейеры, с трансформациями, очисткой и прочей хуйнёй. Для бизнес-аналитики — самое то.
• CDC (Change Data Capture) — вот это, блядь, магия! Инструменты вроде Debezium садятся на лог транзакций исходной базы и в реальном времени ловят все изменения: инсерты, апдейты, делиты. И тут же шлют их в целевую. Получается живая, непрерывная синхронизация. Красота!
• Встроенная репликация — если базы одной масти (два PostgreSQL, например), можно настроить родную репликацию. База сама всё сделает, только смотри да подливай.

А как мигрировать с минимумом простоя, чтобы бизнес не приехал тебе на уши?

1. Сначала историю — переноси все старые, архивные данные, пока система работает.
2. Потом дельту — лови изменения, которые накопились за время переноса истории.
3. Короткая блокировка — на пару секунд останавливаешь запись в старую базу, переносишь последние, самые свежие изменения. Это самый пиздецовый момент, терпения ноль ебать!
4. Переключение — и быстренько, как мартышлюшка, переводишь всех на новую базу.
5. Двунаправленная запись (Dual Write) — это для параноиков. Настраиваешь приложение, чтобы оно писало сразу в обе базы какое-то время. И читаешь со старой. Потом, когда уверен, что новая не глючит, переключаешь чтение на неё. Сложно, зато надёжно, как швейцарские часы.

Вот и вся наука. Главное — не паниковать, тестировать на тестовых стендах до посинения и всегда иметь план отката. А то бывает так перенесёшь, что потом волком вой.