Как сравнить хэш-суммы таблиц при переносе данных между СУБД

При переносе данных из SQLite или SQL Server в PostgreSQL важно убедиться, что информация передана без потерь и искажений. Один из надёжных способов - вычислить и сравнить хэш-суммы (контрольные суммы) исходной и целевой таблиц. В этой статье разберём, как реализовать такую проверку с помощью Python.

Зачем нужен контроль хэш-суммы при миграции БД

Хэш-сумма - это уникальная строка фиксированной длины, которая генерируется на основе содержимого данных. Если хотя бы один бит изменится, хэш будет другим. Сравнение хэшей позволяет:

• Обнаружить потерю строк при передаче
• Выявить дубликаты или пропуски в данных
• Проверить корректность типов данных (например, округление чисел)

Для надёжности лучше считать хэш не для всей таблицы целиком, а для каждой строки отдельно, а затем объединить результаты в общую контрольную сумму.

Алгоритм проверки целостности на Python

Основная идея: отсортировать данные по первичному ключу, конкатенировать все значения строки в одну строку, вычислить MD5 или SHA256 для каждой строки, а затем - итоговый хэш для всей таблицы.

Шаг 1. Получаем данные из исходной СУБД

Для SQLite и SQL Server используем соответствующие драйверы (sqlite3 и pyodbc). Пример для SQLite:

import sqlite3
import hashlib

conn = sqlite3.connect('source.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM my_table ORDER BY id')
rows = cursor.fetchall()

Обязательно сортируйте строки по первичному ключу, иначе порядок может отличаться, и хэши не совпадут.

Шаг 2. Вычисляем хэш-сумму таблицы

Создаём функцию, которая принимает список строк и возвращает один хэш:

def compute_table_hash(rows):
    hasher = hashlib.sha256()
    for row in rows:
        # Преобразуем все поля в строку и объединяем через разделитель
        row_str = '|'.join(str(value) for value in row)
        hasher.update(row_str.encode('utf-8'))
    return hasher.hexdigest()

Разделитель (например, |) нужен, чтобы избежать коллизий, когда значения сливаются воедино.

Шаг 3. Повторяем для целевой таблицы в PostgreSQL

Аналогично подключаемся к PostgreSQL через psycopg2 и получаем те же данные в том же порядке сортировки:

import psycopg2

conn_pg = psycopg2.connect('dbname=target user=...')
cursor_pg = conn_pg.cursor()
cursor_pg.execute('SELECT * FROM my_table ORDER BY id')
rows_pg = cursor_pg.fetchall()

Шаг 4. Сравниваем хэши

Вызываем функцию для обоих наборов данных и сравниваем:

hash_source = compute_table_hash(rows)
hash_target = compute_table_hash(rows_pg)

if hash_source == hash_target:
    print('Данные идентичны. Миграция прошла успешно.')
else:
    print('Ошибка: хэши не совпадают! Требуется проверка.')

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

Советы для разных СУБД

• SQLite: поддерживает только стандартные типы, поэтому конвертация в строку обычно проходит без сюрпризов.
• SQL Server: осторожнее с типами DATETIME и FLOAT. Приводите к строке с фиксированным форматом, чтобы избежать различий из-за локальных настроек.
• PostgreSQL: при использовании ORDER BY учитывайте, что NULL значения могут сортироваться по-разному. Явно задайте NULLS LAST или NULLS FIRST.

Альтернативные методы проверки

Помимо хэширования на стороне Python, можно использовать встроенные функции СУБД. Например, в PostgreSQL есть CHECKSUM для таблиц, но он не гарантирует криптографической стойкости. В SQL Server функция CHECKSUM_AGG работает только для одного столбца. Поэтому ручной расчёт через Python остаётся наиболее универсальным и контролируемым способом.

Заключение

Сравнение хэш-сумм - эффективный метод верификации данных при миграции между SQLite, SQL Server и PostgreSQL. Реализация на Python с использованием hashlib и сортировки по первичному ключу позволяет быстро обнаружить расхождения. Для больших таблиц рекомендуется разбивать проверку на блоки, чтобы локализовать ошибки.