本文目录导读：
迁移脚本编写策略
全量数据导出脚本
-- 导出所有表数据（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
最佳实践建议
- 分批处理：设置合理的批处理大小（500-1000条/批）
- 错误处理：实现重试机制和错误日志
- 断点续传：记录已迁移的数据ID，支持从断点继续
- 数据转换：统一字符集、时区、格式
- 性能优化：
- 禁用目标表的索引和约束
- 使用批量插入
- 增加数据库连接池
- 备份策略：迁移前备份源数据库
需要我详细说明某个特定场景的迁移方案吗？
