本文目录导读：
我来提供一个完整的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
注意事项
- 备份: 迁移前务必备份源数据库
- 测试: 建议先在测试环境验证迁移脚本
- 性能: 设置合适的批次大小，避免内存溢出
- 错误处理: 迁移脚本应该有完善的错误处理和日志记录
- 断点续传: 对大表启用断点续传功能
- 数据验证: 迁移后执行数据完整性检查
这个脚本提供了完整的数据库迁移解决方案,可根据实际需求进行调整和扩展。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
