脚本如何迁移旧数据库数据内容

本文目录导读：

1. 迁移脚本编写策略
2. 自动迁移脚本（Shell）
3. 复杂迁移脚本（Node.js）
4. 数据验证脚本
5. 最佳实践建议

迁移脚本编写策略

全量数据导出脚本

-- 导出所有表数据（MySQL示例）
mysqldump -u username -p old_database > backup.sql
-- 导出指定表
mysqldump -u username -p old_database table1 table2 > partial_backup.sql

增量迁移脚本（Python示例）

import pymysql
import psycopg2
from datetime import datetime
def migrate_data(batch_size=1000):
    # 连接新旧数据库
    old_conn = pymysql.connect(host='old_host', user='old_user', password='old_pass', db='old_db')
    new_conn = psycopg2.connect(host='new_host', user='new_user', password='new_pass', dbname='new_db')
    old_cursor = old_conn.cursor(pymysql.cursors.DictCursor)
    new_cursor = new_conn.cursor()
    # 分批次迁移
    offset = 0
    while True:
        old_cursor.execute(f"SELECT * FROM users LIMIT {batch_size} OFFSET {offset}")
        rows = old_cursor.fetchall()
        if not rows:
            break
        for row in rows:
            # 数据转换逻辑
            transformed = transform_data(row)
            new_cursor.execute("""
                INSERT INTO users (id, name, email, created_at)
                VALUES (%s, %s, %s, %s)
            """, (transformed['id'], transformed['name'], 
                  transformed['email'], transformed['created_at']))
        new_conn.commit()
        offset += batch_size
        print(f"已迁移 {offset} 条记录")
    old_cursor.close()
    new_cursor.close()
    old_conn.close()
    new_conn.close()
def transform_data(row):
    """数据转换示例：处理字段差异"""
    return {
        'id': row['user_id'],
        'name': row['full_name'].strip(),
        'email': row['email'].lower(),
        'created_at': datetime.strptime(row['reg_time'], '%Y-%m-%d %H:%M:%S')
    }

自动迁移脚本（Shell）

#!/bin/bash
# 数据库配置
OLD_DB="mysql://old_user:old_pass@old_host:3306/old_db"
NEW_DB="postgresql://new_user:new_pass@new_host:5432/new_db"
LOG_FILE="migration_$(date +%Y%m%d_%H%M%S).log"
# 迁移函数
migrate_table() {
    table=$1
    echo "[$(date)] 开始迁移表: $table" | tee -a $LOG_FILE
    # 使用管道直接传输
    mysqldump --compatible=postgresql \
              --default-character-set=utf8 \
              --no-create-info \
              --skip-create-options \
              $OLD_DB $table | \
    sed -e 's/`/"/g' \
        -e "s/'/\'\'/g" \
        -e "s/\\\'/\'\'/g" | \
    psql $NEW_DB >> $LOG_FILE 2>&1
    if [ $? -eq 0 ]; then
        echo "[$(date)] 表 $table 迁移成功" | tee -a $LOG_FILE
    else
        echo "[$(date)] 表 $table 迁移失败" | tee -a $LOG_FILE
        return 1
    fi
}
# 主流程
echo "开始数据库迁移 - $(date)" | tee $LOG_FILE
# 迁移所有表
tables=("user" "order" "product" "category")
for table in "${tables[@]}"; do
    migrate_table $table
done
echo "迁移完成 - $(date)" | tee -a $LOG_FILE

复杂迁移脚本（Node.js）

const mysql = require('mysql2/promise');
const { Client } = require('pg');
const { Transform } = require('stream');
const { pipeline } = require('stream/promises');
const ProgressBar = require('progress');
class DatabaseMigrator {
    constructor(config) {
        this.source = config.source;
        this.target = config.target;
        this.batchSize = config.batchSize || 1000;
        this.migrations = [];
    }
    async connect() {
        this.sourceConn = await mysql.createConnection(this.source);
        this.targetConn = new Client(this.target);
        await this.targetConn.connect();
    }
    async migrateTable(tableName, options = {}) {
        const {
            where = '1=1',
            transform = (row) => row,
            onProgress = () => {},
            batchSize = this.batchSize
        } = options;
        console.log(`开始迁移表: ${tableName}`);
        let offset = 0;
        let totalRows = 0;
        // 获取总记录数
        const [countResult] = await this.sourceConn.execute(
            `SELECT COUNT(*) as total FROM ${tableName} WHERE ${where}`
        );
        totalRows = countResult[0].total;
        const progressBar = new ProgressBar(
            `迁移 ${tableName} [:bar] :current/:total :percent :etas`,
            { total: totalRows, width: 40 }
        );
        while (offset < totalRows) {
            // 读取数据
            const [rows] = await this.sourceConn.execute(
                `SELECT * FROM ${tableName} WHERE ${where} LIMIT ? OFFSET ?`,
                [batchSize, offset]
            );
            // 批量插入
            const values = rows.map(row => transform(row));
            await this.batchInsert(tableName, values);
            offset += batchSize;
            progressBar.update(offset / totalRows);
            onProgress({ table: tableName, offset, totalRows });
            // 添加延迟避免锁表
            await this.sleep(100);
        }
        console.log(`\n表 ${tableName} 迁移完成`);
    }
    async batchInsert(tableName, rows) {
        if (rows.length === 0) return;
        const columns = Object.keys(rows[0]);
        const placeholders = rows.map(() => 
            `(${columns.map(() => '$' + (columns.indexOf(columns.find(c => true)) + 1)).join(',')})`
        ).join(',');
        const query = `
            INSERT INTO ${tableName} (${columns.join(',')})
            VALUES ${placeholders}
            ON CONFLICT (id) DO UPDATE SET 
            ${columns.map(col => `${col} = EXCLUDED.${col}`).join(',')}
        `;
        const values = rows.flatMap(row => 
            columns.map(col => row[col])
        );
        await this.targetConn.query(query, values);
    }
    sleep(ms) {
        return new Promise(resolve => setTimeout(resolve, ms));
    }
}
// 使用示例
const migrator = new DatabaseMigrator({
    source: {
        host: 'localhost',
        user: 'root',
        password: 'pass',
        database: 'old_db'
    },
    target: {
        host: 'new_host',
        user: 'new_user',
        password: 'new_pass',
        database: 'new_db'
    },
    batchSize: 500
});
async function main() {
    try {
        await migrator.connect();
        // 迁移用户表并转换数据
        await migrator.migrateTable('users', {
            transform: (row) => ({
                id: row.user_id,
                username: row.user_name,
                email: row.user_email,
                phone: row.phone.replace(/\s/g, ''),
                created_at: new Date(row.register_time),
                status: row.active === 1 ? 'active' : 'inactive'
            })
        });
        // 迁移订单表
        await migrator.migrateTable('orders', {
            where: 'status != "deleted"',
            transform: (row) => ({
                id: row.order_id,
                user_id: row.customer_id,
                total: parseFloat(row.total_amount),
                currency: row.currency.toUpperCase(),
                ordered_at: new Date(row.create_time)
            })
        });
        console.log('所有表迁移完成！');
        await migrator.targetConn.end();
        await migrator.sourceConn.end();
    } catch (error) {
        console.error('迁移失败:', error);
        process.exit(1);
    }
}
main();

数据验证脚本

def verify_migration():
    """验证迁移后的数据完整性"""
    checks = {
        'user_count': "SELECT COUNT(*) FROM users",
        'data_sample': "SELECT id, name, email FROM users LIMIT 10",
        'created_dates': "SELECT MIN(created_at), MAX(created_at) FROM users"
    }
    for name, query in checks.items():
        old_result = execute_old_db(query)
        new_result = execute_new_db(query)
        if old_result != new_result:
            print(f"验证失败: {name}")
            print(f"旧数据库: {old_result}")
            print(f"新数据库: {new_result}")
            return False
    return True

最佳实践建议

1. 分批处理：设置合理的批处理大小（500-1000条/批）
2. 错误处理：实现重试机制和错误日志
3. 断点续传：记录已迁移的数据ID，支持从断点继续
4. 数据转换：统一字符集、时区、格式
5. 性能优化：
   • 禁用目标表的索引和约束
   • 使用批量插入
   • 增加数据库连接池
6. 备份策略：迁移前备份源数据库

需要我详细说明某个特定场景的迁移方案吗？