不同架构新旧数据库全表数据对比方案求助（禁第三方工具/Excel）

Hey there, let's break down how to tackle this large-scale heterogeneous database data comparison challenge without relying on third-party tools or Excel—since those options are off the table for your scenario.

第一步：先梳理表/字段映射关系 #

This is the foundational step—you can't compare data if you don't know which old maps to which new.

• Create a clear mapping table that links every table and field from the old database to its counterpart in the new one. Don't forget to note data type conversions (e.g., old varchar(50) → new string, or date format differences like YYYY/MM/DD vs YYYY-MM-DD).
• Identify unique identifiers: Pin down the primary key or field combination that uniquely identifies a record in both databases (e.g., old user_id → new usr_uuid). This is critical for matching records accurately.

第二步：编写自定义对比脚本（推荐 Python/Shell，适配你的数据库类型） #

Writing your own script lets you handle large datasets efficiently (via batching) and customize the logic to fit your schema differences. Here's a practical Python example for MySQL ↔ PostgreSQL comparison:

Core Logic #

1. Pull data in batches (to avoid memory overload with large datasets)
2. Transform old data to match the new database's field names and data formats
3. Use hashing (like MD5) for fast row-level comparison, or do field-by-field checks if you need to pinpoint exact differences

Sample Code Snippet #

import hashlib
import mysql.connector
import psycopg2

# 1. Connect to both databases
old_db_conn = mysql.connector.connect(
    host="old_db_host",
    user="your_username",
    password="your_password",
    database="old_db_name"
)
new_db_conn = psycopg2.connect(
    host="new_db_host",
    user="your_username",
    password="your_password",
    dbname="new_db_name"
)

# 2. Define field mapping: old_table.field → new_table.field
field_mapping = {
    "old_users.id": "new_users.usr_uuid",
    "old_users.full_name": "new_users.display_name",
    "old_users.join_date": "new_users.signup_timestamp"
}

# 3. Batch processing to handle large data volumes
batch_size = 1000
offset = 0

while True:
    # Fetch batch from old database
    old_cursor = old_db_conn.cursor(dictionary=True)
    old_cursor.execute(
        f"SELECT id, full_name, join_date FROM old_users LIMIT {batch_size} OFFSET {offset}"
    )
    old_records = old_cursor.fetchall()
    
    if not old_records:
        break  # Exit loop when no more records
    
    # Transform old data and generate row hashes
    old_record_hashes = {}
    for record in old_records:
        unique_key = record["id"]
        # Normalize data format (e.g., standardize date strings)
        transformed_data = {
            field_mapping["old_users.full_name"]: record["full_name"].strip(),
            field_mapping["old_users.join_date"]: record["join_date"].strftime("%Y-%m-%d %H:%M:%S")
        }
        # Create hash for quick comparison
        row_hash = hashlib.md5(str(transformed_data).encode()).hexdigest()
        old_record_hashes[unique_key] = row_hash
    
    # Fetch corresponding records from new database
    new_cursor = new_db_conn.cursor(dictionary=True)
    placeholders = ','.join(['%s'] * len(old_record_hashes.keys()))
    new_cursor.execute(
        f"SELECT usr_uuid, display_name, signup_timestamp FROM new_users WHERE usr_uuid IN ({placeholders})",
        tuple(old_record_hashes.keys())
    )
    new_records = new_cursor.fetchall()
    
    # Compare hashes and flag mismatches
    for record in new_records:
        unique_key = record["usr_uuid"]
        transformed_new_data = {
            field_mapping["old_users.full_name"]: record["display_name"].strip(),
            field_mapping["old_users.join_date"]: record["signup_timestamp"].strftime("%Y-%m-%d %H:%M:%S")
        }
        new_row_hash = hashlib.md5(str(transformed_new_data).encode()).hexdigest()
        
        if old_record_hashes[unique_key] != new_row_hash:
            print(f"Mismatch detected for record ID: {unique_key}")
            # Uncomment below to get field-level differences
            # for field in transformed_new_data:
            #     if transformed_old_data[field] != transformed_new_data[field]:
            #         print(f"Field {field}: Old value = {transformed_old_data[field]}, New value = {transformed_new_data[field]}")
    
    offset += batch_size

# Clean up connections
old_db_conn.close()
new_db_conn.close()

第三步：Handle Edge Cases #

• Ultra-large datasets (10M+ records)：Generate hashes directly in the database (e.g., MySQL's MD5(CONCAT_WS('|', id, full_name, join_date))) and export these hash lists to text files. Use the diff command (Linux) or PowerShell's Compare-Object (Windows) to compare the files—this is way faster than processing in code.
• Missing/extra records：Extract the full list of unique identifiers from both databases, then compute the difference to find records that exist in one but not the other.
• Data type precision issues：For numeric fields (e.g., old DECIMAL(10,2) → new FLOAT), normalize values to the same precision before comparison (e.g., convert to strings with 2 decimal places).

第四步： Validate Results #

After running the script, manually spot-check a subset of flagged mismatches to ensure your mapping and transformation logic is correct—this avoids false positives from schema mapping errors.

内容的提问来源于stack exchange，提问作者arpit goyal