本文目录导读：
我来提供一个完整的PHP历史数据迁移脚本示例,包含多种常见场景。
基础数据库迁移脚本
<?php
/**
* PHP数据库历史数据迁移脚本
*/
class DatabaseMigrator {
private $sourcePDO;
private $targetPDO;
private $batchSize = 1000;
private $logFile;
public function __construct($sourceConfig, $targetConfig) {
$this->sourcePDO = $this->createConnection($sourceConfig);
$this->targetPDO = $this->createConnection($targetConfig);
$this->logFile = __DIR__ . '/migration_' . date('Y-m-d_H-i-s') . '.log';
}
// 创建数据库连接
private function createConnection($config) {
$dsn = "mysql:host={$config['host']};dbname={$config['database']};charset=utf8mb4";
$options = [
PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
PDO::ATTR_EMULATE_PREPARES => false
];
try {
return new PDO($dsn, $config['username'], $config['password'], $options);
} catch (PDOException $e) {
$this->log("连接失败: " . $e->getMessage(), 'ERROR');
throw $e;
}
}
// 日志记录
private function log($message, $level = 'INFO') {
$logEntry = sprintf("[%s] [%s] %s\n", date('Y-m-d H:i:s'), $level, $message);
file_put_contents($this->logFile, $logEntry, FILE_APPEND);
echo $logEntry;
}
// 单表迁移
public function migrateTable($tableName, callable $transform = null) {
try {
$this->log("开始迁移表: {$tableName}");
// 获取源表总记录数
$countQuery = "SELECT COUNT(*) FROM {$tableName}";
$stmt = $this->sourcePDO->query($countQuery);
$totalCount = $stmt->fetchColumn();
$this->log("源表记录总数: {$totalCount}");
// 分批次读取和插入
$offset = 0;
$migratedCount = 0;
while ($offset < $totalCount) {
// 从源表读取数据
$selectStmt = $this->sourcePDO->prepare(
"SELECT * FROM {$tableName} LIMIT :limit OFFSET :offset"
);
$selectStmt->bindValue(':limit', $this->batchSize, PDO::PARAM_INT);
$selectStmt->bindValue(':offset', $offset, PDO::PARAM_INT);
$selectStmt->execute();
$rows = $selectStmt->fetchAll();
if (empty($rows)) {
break;
}
// 如果提供了转换函数，处理数据
if ($transform) {
$rows = array_map($transform, $rows);
}
// 批量插入到目标表
$this->batchInsert($tableName, $rows);
$migratedCount += count($rows);
$offset += $this->batchSize;
$this->log("进度: {$migratedCount}/{$totalCount}");
}
$this->log("表 {$tableName} 迁移完成，共迁移 {$migratedCount} 条记录");
return $migratedCount;
} catch (Exception $e) {
$this->log("迁移表 {$tableName} 失败: " . $e->getMessage(), 'ERROR');
throw $e;
}
}
// 批量插入数据
private function batchInsert($tableName, $rows) {
if (empty($rows)) {
return;
}
$columns = array_keys($rows[0]);
$columnList = implode(', ', $columns);
$placeholders = '(' . implode(', ', array_fill(0, count($columns), '?')) . ')';
$sql = "INSERT INTO {$tableName} ({$columnList}) VALUES " .
implode(', ', array_fill(0, count($rows), $placeholders));
try {
$this->targetPDO->beginTransaction();
$stmt = $this->targetPDO->prepare($sql);
$parameters = [];
foreach ($rows as $row) {
foreach ($columns as $column) {
$parameters[] = $row[$column];
}
}
$stmt->execute($parameters);
$this->targetPDO->commit();
} catch (Exception $e) {
$this->targetPDO->rollBack();
$this->log("批量插入失败: " . $e->getMessage(), 'ERROR');
throw $e;
}
}
// 复杂迁移示例：关联表数据迁移
public function migrateComplexData() {
$this->log("开始复杂数据迁移");
try {
// 从源表读取
$sql = "
SELECT
o.id as old_order_id,
o.order_no,
o.customer_id,
o.amount,
o.created_at,
c.name as customer_name,
c.email as customer_email
FROM old_orders o
LEFT JOIN old_customers c ON o.customer_id = c.id
WHERE o.created_at < '2023-01-01'
ORDER BY o.created_at ASC
";
$stmt = $this->sourcePDO->query($sql);
// 目标表插入
$insertSql = "
INSERT INTO new_orders (
order_no, customer_id, amount,
customer_name, customer_email, created_at,
migrated_at, migration_batch
) VALUES (
:order_no, :customer_id, :amount,
:customer_name, :customer_email, :created_at,
NOW(), :batch
)
";
$insertStmt = $this->targetPDO->prepare($insertSql);
$this->targetPDO->beginTransaction();
$batch = date('YmdHis');
$count = 0;
while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
// 数据转换处理
$row['amount'] = (float)$row['amount'];
$row['created_at'] = date('Y-m-d H:i:s', strtotime($row['created_at']));
$insertStmt->execute([
':order_no' => $row['order_no'],
':customer_id' => $row['customer_id'],
':amount' => $row['amount'],
':customer_name' => $row['customer_name'],
':customer_email' => $row['customer_email'],
':created_at' => $row['created_at'],
':batch' => $batch
]);
$count++;
// 每1000条提交一次
if ($count % 1000 == 0) {
$this->targetPDO->commit();
$this->targetPDO->beginTransaction();
$this->log("已迁移 {$count} 条复杂数据");
// 源表标记已迁移
$this->markAsMigrated($row['old_order_id']);
}
}
$this->targetPDO->commit();
$this->log("复杂数据迁移完成，共迁移 {$count} 条记录");
} catch (Exception $e) {
if ($this->targetPDO->inTransaction()) {
$this->targetPDO->rollBack();
}
$this->log("复杂数据迁移失败: " . $e->getMessage(), 'ERROR');
throw $e;
}
}
// 标记数据已迁移
private function markAsMigrated($id) {
try {
$stmt = $this->sourcePDO->prepare(
"UPDATE old_orders SET migrated = 1 WHERE id = ?"
);
$stmt->execute([$id]);
} catch (Exception $e) {
$this->log("标记迁移状态失败: " . $e->getMessage(), 'WARNING');
}
}
// 数据完整性验证
public function validateMigration($tableName) {
$this->log("开始验证表 {$tableName} 的数据完整性");
try {
// 对比记录数
$sourceCount = $this->sourcePDO->query("SELECT COUNT(*) FROM {$tableName}")->fetchColumn();
$targetCount = $this->targetPDO->query("SELECT COUNT(*) FROM {$tableName}")->fetchColumn();
$match = ($sourceCount == $targetCount);
$this->log("记录数对比 - 源表: {$sourceCount}, 目标表: {$targetCount}, 匹配: " . ($match ? '是' : '否'));
// 如果数量不匹配，进行详细检查
if (!$match) {
$sampleData = $this->sampleCheck($tableName);
return $sampleData;
}
return ['source_count' => $sourceCount, 'target_count' => $targetCount, 'match' => true];
} catch (Exception $e) {
$this->log("验证失败: " . $e->getMessage(), 'ERROR');
return false;
}
}
// 抽样检查
private function sampleCheck($tableName, $sampleSize = 100) {
$this->log("进行抽样检查...");
// 从源表取样本
$sourceSample = $this->sourcePDO->query(
"SELECT * FROM {$tableName} ORDER BY RAND() LIMIT {$sampleSize}"
)->fetchAll();
$mismatches = [];
foreach ($sourceSample as $row) {
$id = $row['id'];
$targetCheck = $this->targetPDO->prepare(
"SELECT * FROM {$tableName} WHERE id = ?"
);
$targetCheck->execute([$id]);
$targetRow = $targetCheck->fetch();
if ($targetRow != $row) {
$mismatches[] = [
'id' => $id,
'source' => $row,
'target' => $targetRow
];
}
}
$this->log("抽样检查完成，发现 " . count($mismatches) . " 个不匹配项");
return $mismatches;
}
}
独立的迁移执行脚本
<?php
/**
* 独立的迁移执行脚本
*/
require_once 'DatabaseMigrator.php';
// 配置
$sourceConfig = [
'host' => 'old-db-host',
'database' => 'old_database',
'username' => 'old_user',
'password' => 'old_password'
];
$targetConfig = [
'host' => 'new-db-host',
'database' => 'new_database',
'username' => 'new_user',
'password' => 'new_password'
];
// 创建迁移实例
$migrator = new DatabaseMigrator($sourceConfig, $targetConfig);
// 执行迁移
try {
echo "=== 开始数据库迁移 ===\n\n";
// 1. 基本表迁移
echo "1. 迁移用户表...\n";
$userCount = $migrator->migrateTable('users');
echo "用户表迁移完成: {$userCount} 条记录\n\n";
echo "2. 迁移订单表...\n";
$orderCount = $migrator->migrateTable('orders', function($row) {
// 数据转换
if (isset($row['status'])) {
$statusMap = [
'1' => 'pending',
'2' => 'processing',
'3' => 'completed',
'4' => 'cancelled'
];
$row['status'] = isset($statusMap[$row['status']]) ?
$statusMap[$row['status']] : 'unknown';
}
return $row;
});
echo "订单表迁移完成: {$orderCount} 条记录\n\n";
// 2. 复杂数据迁移
echo "3. 执行复杂数据迁移...\n";
$migrator->migrateComplexData();
echo "复杂数据迁移完成\n\n";
// 3. 数据验证
echo "4. 验证数据完整性...\n";
$validationResult = $migrator->validateMigration('users');
print_r($validationResult);
echo "\n=== 数据库迁移完成 ===\n";
} catch (Exception $e) {
echo "迁移失败: " . $e->getMessage() . "\n";
exit(1);
}
带恢复机制的迁移脚本
<?php
/**
* 带检查点和恢复机制的迁移脚本
*/
class AdvancedMigrator extends DatabaseMigrator {
private $checkpointFile;
private $checkpoint;
public function __construct($sourceConfig, $targetConfig, $checkpointFile = null) {
parent::__construct($sourceConfig, $targetConfig);
$this->checkpointFile = $checkpointFile ?? __DIR__ . '/migration_checkpoint.json';
$this->loadCheckpoint();
}
// 加载检查点
private function loadCheckpoint() {
if (file_exists($this->checkpointFile)) {
$json = file_get_contents($this->checkpointFile);
$this->checkpoint = json_decode($json, true);
} else {
$this->checkpoint = [];
}
}
// 保存检查点
private function saveCheckpoint() {
file_put_contents($this->checkpointFile, json_encode($this->checkpoint, JSON_PRETTY_PRINT));
}
// 记录迁移进度
private function recordProgress($task, $completed, $total, $extra = []) {
$this->checkpoint[$task] = [
'completed' => $completed,
'total' => $total,
'timestamp' => date('Y-m-d H:i:s'),
'extra' => $extra
];
$this->saveCheckpoint();
}
// 执行可恢复的迁移
public function migrateWithResume($tableName) {
$taskKey = "table_{$tableName}";
// 如果已有检查点，从上次位置继续
if (isset($this->checkpoint[$taskKey])) {
$lastPosition = $this->checkpoint[$taskKey]['completed'];
$this->log("从上次位置恢复: {$lastPosition}");
return $this->resumeTableMigration($tableName, $lastPosition);
}
// 全新迁移
try {
$this->log("开始迁移表: {$tableName}");
$countQuery = "SELECT COUNT(*) FROM {$tableName}";
$totalCount = $this->sourcePDO->query($countQuery)->fetchColumn();
// 如果是第一次，记录总记录数
if (!isset($this->checkpoint[$taskKey])) {
$this->recordProgress($taskKey, 0, $totalCount);
}
$offset = 0;
$transferred = 0;
while ($transferred < $totalCount) {
// 检查是否需要恢复
if (isset($this->checkpoint[$taskKey]['completed'])) {
$offset = $this->checkpoint[$taskKey]['completed'];
unset($this->checkpoint[$taskKey]['completed']); // 清除恢复标记
}
// 读取和处理数据
$selectStmt = $this->sourcePDO->prepare(
"SELECT * FROM {$tableName} LIMIT :limit OFFSET :offset"
);
$selectStmt->bindValue(':limit', $this->batchSize, PDO::PARAM_INT);
$selectStmt->bindValue(':offset', $offset, PDO::PARAM_INT);
$selectStmt->execute();
$rows = $selectStmt->fetchAll();
if (empty($rows)) break;
// 批量插入
$this->batchInsert($tableName, $rows);
$offset += count($rows);
$transferred += count($rows);
// 更新进度
$this->recordProgress($taskKey, $offset, $totalCount);
$this->log("进度: {$offset}/{$totalCount}");
}
$this->log("表 {$tableName} 迁移完成");
return true;
} catch (Exception $e) {
$this->log("迁移中断: " . $e->getMessage(), 'ERROR');
$this->log("可以在下次运行时自动从断点恢复", 'INFO');
throw $e;
}
}
// 断点续传函数
private function resumeTableMigration($tableName, $position) {
return $this->migrateWithResume($tableName);
}
}
// 使用示例
$config = [
'source' => [
'host' => 'localhost',
'database' => 'old_db',
'username' => 'user',
'password' => 'pass'
],
'target' => [
'host' => 'localhost',
'database' => 'new_db',
'username' => 'user',
'password' => 'pass'
]
];
$migrator = new AdvancedMigrator($config['source'], $config['target']);
$migrator->migrateWithResume('users');
命令行迁移工具
<?php
/**
* 命令行迁移脚本工具
* 使用方法: php migrate.php --table=table_name [options]
*/
require_once 'DatabaseMigrator.php';
// 命令行参数解析
$options = getopt('', ['table:', 'source:', 'target:', 'help']);
function showHelp() {
echo <<<HELP
PHP数据迁移工具
用法: php migrate.php [options]
选项:
--table=TABLE 指定要迁移的表名
--source=CONFIG 源数据库配置(JSON格式)
--target=CONFIG 目标数据库配置(JSON格式)
--help 显示帮助信息
示例:
php migrate.php --table=orders
php migrate.php --table=users --source='{"host":"localhost","database":"db1"}'
HELP;
}
if (isset($options['help'])) {
showHelp();
exit(0);
}
// 加载配置
$configFile = __DIR__ . '/config.php';
if (file_exists($configFile)) {
$config = require $configFile;
} else {
$config = [
'source' => json_decode($options['source'] ?? '{}', true),
'target' => json_decode($options['target'] ?? '{}', true)
];
}
try {
$migrator = new DatabaseMigrator($config['source'], $config['target']);
if (isset($options['table'])) {
// 迁移单个表
$result = $migrator->migrateTable($options['table']);
echo "迁移完成: {$result} 条记录\n";
} else {
// 迁移所有表
$tables = ['users', 'orders', 'products', 'categories'];
foreach ($tables as $table) {
$migrator->migrateTable($table);
}
}
} catch (Exception $e) {
echo "错误: " . $e->getMessage() . "\n";
exit(1);
}
配置文件示例
<?php
// config.php - 迁移配置
return [
'source' => [
'host' => env('OLD_DB_HOST', 'localhost'),
'database' => env('OLD_DB_DATABASE', 'old_database'),
'username' => env('OLD_DB_USERNAME', 'old_user'),
'password' => env('OLD_DB_PASSWORD', 'old_password'),
],
'target' => [
'host' => env('NEW_DB_HOST', 'localhost'),
'database' => env('NEW_DB_DATABASE', 'new_database'),
'username' => env('NEW_DB_USERNAME', 'new_user'),
'password' => env('NEW_DB_PASSWORD', 'new_password'),
],
'batch_size' => 1000,
'tables' => [
'users' => [
'transform' => function($row) {
// 添加迁移标记
$row['migrated_at'] = date('Y-m-d H:i:s');
$row['migration_batch'] = getenv('MIGRATION_BATCH');
return $row;
}
],
'orders' => [
'conditions' => "WHERE created_at >= '2023-01-01'",
'transform' => function($row) {
// 状态转换
$statusMap = ['pending', 'processing', 'completed'];
$row['status'] = isset($statusMap[$row['status_id']]) ?
$statusMap[$row['status_id']] : 'unknown';
return $row;
}
]
]
];
使用环境变量
// .env 文件示例 OLD_DB_HOST=localhost OLD_DB_DATABASE=old_database OLD_DB_USERNAME=root OLD_DB_PASSWORD=secret NEW_DB_HOST=localhost NEW_DB_DATABASE=new_database NEW_DB_USERNAME=root NEW_DB_PASSWORD=newsecret MIGRATION_BATCH=20240101
运行方式
# 运行迁移
php migrate.php --table=users
# 带自定义配置
php migrate.php --table=orders --source='{"host":"192.168.1.100","database":"old_db"}'
# 执行完整迁移
php migrate.php --all
# 带调试输出
php -d display_errors=1 migrate.php --table=users
注意事项
- 备份: 迁移前务必备份源数据库
- 测试: 建议先在测试环境验证迁移脚本
- 性能: 设置合适的批次大小，避免内存溢出
- 错误处理: 迁移脚本应该有完善的错误处理和日志记录
- 断点续传: 对大表启用断点续传功能
- 数据验证: 迁移后执行数据完整性检查
这个脚本提供了完整的数据库迁移解决方案,可根据实际需求进行调整和扩展。
