引言
数据库迁移是软件开发和维护过程中常见的任务，尤其是当项目从小规模发展到大规模时，从轻量级的SQLite迁移到功能更强大的MySQL是一个常见的需求。本文将详细介绍如何从SQLite平滑迁移到MySQL，解决迁移过程中可能遇到的数据转换、表结构差异以及性能优化等问题，帮助您顺利完成数据库迁移工作。
SQLite与MySQL的对比
SQLite特点
SQLite是一个轻量级的嵌入式数据库，具有以下特点：
无需服务器，直接访问数据库文件
零配置，无需安装和管理
适合小型应用、移动应用和单用户桌面应用
支持大多数SQL标准，但有一些功能限制
整个数据库存储在单个文件中
MySQL特点
MySQL是一个功能强大的关系型数据库管理系统，具有以下特点：
客户端/服务器架构
支持多用户同时访问
提供完善的安全性和权限管理
支持存储过程、触发器、视图等高级功能
更好的性能和扩展性，适合中大型应用
支持多种存储引擎（如InnoDB、MyISAM等）
迁移前的准备工作
评估现有数据库
在开始迁移之前，需要对现有的SQLite数据库进行全面评估：
数据库大小分析
：统计数据库文件大小、表数量、记录数量等
数据类型检查
：记录SQLite中使用的所有数据类型
依赖关系分析
：分析表之间的关系、外键约束等
特殊功能使用情况
：检查是否使用了SQLite特有的功能
准备MySQL环境
安装MySQL服务器
：根据操作系统选择合适的MySQL版本并安装
创建数据库和用户
：为目标数据库创建专用的数据库和用户账号
配置MySQL参数
：根据应用需求调整MySQL配置参数
— 创建数据库
CREATE DATABASE your_database_name CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
— 创建用户并授权
CREATE USER ‘your_username’@’localhost’ IDENTIFIED BY ‘your_password’;
GRANT ALL PRIVILEGES ON your_database_name.* TO ‘your_username’@’localhost’;
FLUSH PRIVILEGES;
准备迁移工具
选择合适的迁移工具可以大大简化迁移过程：
SQLite命令行工具
：用于导出SQLite数据库结构和数据
MySQL命令行工具
：用于导入数据到MySQL
第三方迁移工具
：如DBConvert、SQLite to MySQL converter等
自定义脚本
：根据需要编写Python、PHP等脚本辅助迁移
数据库结构迁移
表结构差异分析
SQLite和MySQL在表结构上存在一些差异，需要特别注意：
数据类型差异
：
SQLite使用动态类型系统，而MySQL使用静态类型系统
某些数据类型在两者中的名称和行为不同
自增字段
：
SQLite使用
INTEGER PRIMARY KEY
或
AUTOINCREMENT
MySQL使用
AUTO_INCREMENT
外键约束
：
SQLite默认不启用外键约束，需要手动启用
MySQL默认支持外键约束（使用InnoDB引擎）
数据类型映射
创建SQLite到MySQL的数据类型映射表：
SQLite类型
MySQL类型
说明
INTEGER
INT
整数类型
TEXT
VARCHAR或TEXT
根据文本长度选择
REAL
FLOAT或DOUBLE
浮点数类型
BLOB
BLOB
二进制数据
NUMERIC
DECIMAL
精确数值类型
BOOLEAN
TINYINT(1)
布尔值
创建MySQL表结构
基于SQLite表结构创建MySQL表结构，考虑以下因素：
主键处理
：
“`sql
– SQLite
CREATE TABLE users (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE
);
– MySQL
CREATE TABLE users (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
email VARCHAR(255) UNIQUE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
2. **外键处理**：
“`sql
— SQLite
CREATE TABLE orders (
id INTEGER PRIMARY KEY,
user_id INTEGER,
order_date TEXT,
FOREIGN KEY (user_id) REFERENCES users(id)
);
— MySQL
CREATE TABLE orders (
id INT AUTO_INCREMENT PRIMARY KEY,
user_id INT,
order_date DATE,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
索引处理
：
“`sql
– SQLite
CREATE INDEX idx_users_email ON users(email);
– MySQL
CREATE INDEX idx_users_email ON users(email);
### 自动生成MySQL表结构
可以使用脚本自动将SQLite表结构转换为MySQL表结构：
“`python
import sqlite3
import re
def convert_sqlite_to_mysql_schema(sqlite_db_path, output_file):
# 连接到SQLite数据库
sqlite_conn = sqlite3.connect(sqlite_db_path)
sqlite_cursor = sqlite_conn.cursor()
# 获取所有表名
sqlite_cursor.execute(“SELECT name FROM sqlite_master WHERE type=’table’;”)
tables = sqlite_cursor.fetchall()
with open(output_file, ‘w’) as f:
for table in tables:
table_name = table[0]
# 获取表创建语句
sqlite_cursor.execute(f”SELECT sql FROM sqlite_master WHERE type=’table’ AND name='{table_name}’;”)
create_stmt = sqlite_cursor.fetchone()[0]
# 转换数据类型
create_stmt = re.sub(r’\bINTEGER PRIMARY KEY\b’, ‘INT AUTO_INCREMENT PRIMARY KEY’, create_stmt)
create_stmt = re.sub(r’\bTEXT\b’, ‘VARCHAR(255)’, create_stmt)
create_stmt = re.sub(r’\bREAL\b’, ‘DOUBLE’, create_stmt)
create_stmt = re.sub(r’\bBOOLEAN\b’, ‘TINYINT(1)’, create_stmt)
# 添加MySQL引擎和字符集
create_stmt = create_stmt.rstrip(‘;’) + ‘ ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;’
f.write(f”– 转换表: {table_name}\n”)
f.write(f”{create_stmt}\n\n”)
sqlite_conn.close()
print(f”MySQL表结构已保存到 {output_file}”)
# 使用示例
convert_sqlite_to_mysql_schema(‘database.sqlite’, ‘mysql_schema.sql’)
数据迁移
导出SQLite数据
有几种方法可以从SQLite导出数据：
使用SQLite命令行工具
：
“`bash
导出整个数据库为SQL文件
sqlite3 database.sqlite .dump > database_dump.sql
# 导出特定表的数据
sqlite3 database.sqlite “.dump users” > users_data.sql
2. **使用CSV格式导出**：
“`bash
sqlite3 -header -csv database.sqlite “SELECT * FROM users;” > users.csv
使用Python脚本导出
：
“`python
import sqlite3
import csv
def export_table_to_csv(sqlite_db_path, table_name, output_file):
conn = sqlite3.connect(sqlite_db_path)
cursor = conn.cursor()
cursor.execute(f”SELECT * FROM {table_name}”)
rows = cursor.fetchall()
# 获取列名
column_names = [description[0] for description in cursor.description]
with open(output_file, ‘w’, newline=”, encoding=’utf-8′) as f:
writer = csv.writer(f)
writer.writerow(column_names) # 写入列名
writer.writerows(rows) # 写入数据
conn.close()
print(f”表 {table_name} 已导出到 {output_file}”)
# 使用示例
export_table_to_csv(‘database.sqlite’, ‘users’, ‘users.csv’)
### 转换数据格式
SQLite和MySQL之间的数据格式可能需要转换：
1. **日期时间格式**：
– SQLite通常以文本格式存储日期时间
– MySQL有专门的日期时间类型（DATE, TIME, DATETIME, TIMESTAMP）
2. **布尔值**：
– SQLite使用0和1表示布尔值
– MySQL可以使用TINYINT(1)或BOOL类型
3. **特殊字符处理**：
– 确保正确处理引号、换行符等特殊字符
### 导入数据到MySQL
有几种方法可以将数据导入到MySQL：
1. **使用MySQL命令行工具**：
“`bash
# 导入SQL文件
mysql -u username -p database_name < mysql_schema.sql
mysql -u username -p database_name < data.sql
# 导入CSV文件
mysql -u username -p -e "
LOAD DATA LOCAL INFILE 'users.csv'
INTO TABLE users
FIELDS TERMINATED BY ','
OPTIONALLY ENCLOSED BY '\"'
LINES TERMINATED BY '\n'
IGNORE 1 ROWS;
" database_name
使用Python脚本导入
：
“`python
import mysql.connector
import csv
def import_csv_to_mysql(csv_file, table_name, mysql_config):
conn = mysql.connector.connect(**mysql_config)
cursor = conn.cursor()
with open(csv_file, 'r', encoding='utf-8') as f:
reader = csv.reader(f)
header = next(reader) # 读取列名
# 构建INSERT语句
placeholders = ', '.join(['%s'] * len(header))
columns = ', '.join(header)
sql = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
# 批量插入数据
batch_size = 1000
batch = []
for row in reader:
batch.append(row)
if len(batch) >= batch_size:
cursor.executemany(sql, batch)
conn.commit()
batch = []
# 插入剩余数据
if batch:
cursor.executemany(sql, batch)
conn.commit()
cursor.close()
conn.close()
print(f”数据已从 {csv_file} 导入到表 {table_name}”)
# MySQL配置
mysql_config = {
‘host’: ‘localhost’,
‘user’: ‘your_username’,
‘password’: ‘your_password’,
‘database’: ‘your_database’
}
# 使用示例
import_csv_to_mysql(‘users.csv’, ‘users’, mysql_config)
### 处理大数据量迁移
对于大型数据库，可能需要特殊处理：
1. **分批处理**：
“`python
def migrate_large_table(sqlite_db_path, table_name, mysql_config, batch_size=10000):
sqlite_conn = sqlite3.connect(sqlite_db_path)
sqlite_cursor = sqlite_conn.cursor()
mysql_conn = mysql.connector.connect(**mysql_config)
mysql_cursor = mysql_conn.cursor()
# 获取总记录数
sqlite_cursor.execute(f”SELECT COUNT(*) FROM {table_name}”)
total_records = sqlite_cursor.fetchone()[0]
print(f”表 {table_name} 共有 {total_records} 条记录”)
# 获取列名
sqlite_cursor.execute(f”SELECT * FROM {table_name} LIMIT 1″)
column_names = [description[0] for description in sqlite_cursor.description]
# 构建INSERT语句
placeholders = ‘, ‘.join([‘%s’] * len(column_names))
columns = ‘, ‘.join(column_names)
sql = f”INSERT INTO {table_name} ({columns}) VALUES ({placeholders})”
# 分批读取和插入
offset = 0
while offset < total_records:
sqlite_cursor.execute(f"SELECT * FROM {table_name} LIMIT {batch_size} OFFSET {offset}")
batch = sqlite_cursor.fetchall()
mysql_cursor.executemany(sql, batch)
mysql_conn.commit()
offset += batch_size
print(f"已处理 {min(offset, total_records)}/{total_records} 条记录")
sqlite_conn.close()
mysql_conn.close()
print(f"表 {table_name} 迁移完成")
# 使用示例
migrate_large_table('database.sqlite', 'large_table', mysql_config)
使用MySQL的LOAD DATA INFILE
：
“`python
def export_sqlite_to_csv_for_mysql(sqlite_db_path, table_name, output_file):
conn = sqlite3.connect(sqlite_db_path)
cursor = conn.cursor()
# 获取列名
cursor.execute(f”SELECT * FROM {table_name} LIMIT 1”)
column_names = [description[0] for description in cursor.description]
# 导出数据
with open(output_file, ‘w’, encoding=‘utf-8’, newline=“) as f:
writer = csv.writer(f)
# 写入数据（不包含列名，因为LOAD DATA INFILE不需要）
for row in cursor.execute(f"SELECT * FROM {table_name}"):
writer.writerow(row)
conn.close()
print(f”表 {table_name} 已导出到 {output_file}，适用于MySQL的LOAD DATA INFILE”)
def load_csv_to_mysql(csv_file, table_name, mysql_config):
conn = mysql.connector.connect(**mysql_config)
cursor = conn.cursor()
# 使用LOAD DATA INFILE
cursor.execute(f"""
LOAD DATA LOCAL INFILE '{csv_file}'
INTO TABLE {table_name}
FIELDS TERMINATED BY ','
OPTIONALLY ENCLOSED BY '"'
LINES TERMINATED BY '\n'
""")
conn.commit()
conn.close()
print(f"数据已从 {csv_file} 加载到表 {table_name}")
# 使用示例
export_sqlite_to_csv_for_mysql(‘database.sqlite’, ‘large_table’, ‘large_table.csv’)
load_csv_to_mysql(‘large_table.csv’, ‘large_table’, mysql_config)
## 迁移后的验证和优化
### 数据验证
迁移完成后，需要验证数据的完整性和一致性：
1. **记录数验证**：
```python
def verify_record_counts(sqlite_db_path, mysql_config):
# 连接到SQLite
sqlite_conn = sqlite3.connect(sqlite_db_path)
sqlite_cursor = sqlite_conn.cursor()
# 获取SQLite中的所有表
sqlite_cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = sqlite_cursor.fetchall()
# 连接到MySQL
mysql_conn = mysql.connector.connect(**mysql_config)
mysql_cursor = mysql_conn.cursor()
# 比较每个表的记录数
for table in tables:
table_name = table[0]
# 获取SQLite中的记录数
sqlite_cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
sqlite_count = sqlite_cursor.fetchone()[0]
# 获取MySQL中的记录数
mysql_cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
mysql_count = mysql_cursor.fetchone()[0]
if sqlite_count == mysql_count:
print(f"表 {table_name}: 记录数一致 ({sqlite_count})")
else:
print(f"表 {table_name}: 记录数不一致 (SQLite: {sqlite_count}, MySQL: {mysql_count})")
sqlite_conn.close()
mysql_conn.close()
# 使用示例
verify_record_counts('database.sqlite', mysql_config)
数据抽样验证
：
“`python
def verify_data_samples(sqlite_db_path, mysql_config, sample_size=10):
# 连接到SQLite和MySQL
sqlite_conn = sqlite3.connect(sqlite_db_path)
sqlite_cursor = sqlite_conn.cursor()
mysql_conn = mysql.connector.connect(**mysql_config)
mysql_cursor = mysql_conn.cursor()
# 获取SQLite中的所有表
sqlite_cursor.execute(“SELECT name FROM sqlite_master WHERE type=‘table’;”)
tables = sqlite_cursor.fetchall()
# 对每个表进行抽样验证
for table in tables:
table_name = table[0]
# 获取表结构
sqlite_cursor.execute(f"PRAGMA table_info({table_name})")
columns = [column[1] for column in sqlite_cursor.fetchall()]
column_list = ', '.join(columns)
# 从SQLite获取随机样本
sqlite_cursor.execute(f"SELECT {column_list} FROM {table_name} ORDER BY RANDOM() LIMIT {sample_size}")
sqlite_samples = sqlite_cursor.fetchall()
# 从MySQL获取相同的样本（假设有id列）
if 'id' in columns:
ids = [sample[columns.index('id')] for sample in sqlite_samples]
ids_str = ', '.join(str(id) for id in ids)
mysql_cursor.execute(f"SELECT {column_list} FROM {table_name} WHERE id IN ({ids_str})")
mysql_samples = mysql_cursor.fetchall()
# 比较样本数据
for i, sqlite_sample in enumerate(sqlite_samples):
mysql_sample = mysql_samples[i]
for j, value in enumerate(sqlite_sample):
if value != mysql_sample[j]:
print(f"表 {table_name}, ID {ids[i]}, 列 {columns[j]}: 数据不匹配 (SQLite: {value}, MySQL: {mysql_sample[j]})")
sqlite_conn.close()
mysql_conn.close()
print(“数据抽样验证完成”)
# 使用示例
verify_data_samples(‘database.sqlite’, mysql_config)
### 索引优化
迁移完成后，需要优化索引以提高查询性能：
1. **分析查询模式**：
```sql
-- 在MySQL中启用慢查询日志
SET GLOBAL slow_query_log = 'ON';
SET GLOBAL long_query_time = 1; -- 记录执行时间超过1秒的查询
SET GLOBAL slow_query_log_file = '/var/log/mysql/mysql-slow.log';
添加必要的索引
：
“`sql
– 为常用查询条件添加索引
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_orders_user_id ON orders(user_id);
CREATE INDEX idx_orders_date ON orders(order_date);
– 为复合查询条件添加复合索引
CREATE INDEX idx_orders_user_date ON orders(user_id, order_date);
3. **删除不必要的索引**：
```sql
-- 查看索引使用情况
SHOW INDEX FROM users;
-- 删除未使用的索引
DROP INDEX idx_unused ON table_name;
查询优化
优化SQL查询以适应MySQL：
使用EXPLAIN分析查询
：
EXPLAIN SELECT * FROM users WHERE email = 'user@example.com';
优化JOIN操作
：
“`sql
– 优化前的查询
SELECT u.name, o.order_date, o.total
FROM users u
JOIN orders o ON u.id = o.user_id
WHERE u.status = ‘active’;
– 优化后的查询（确保JOIN字段有索引）
SELECT u.name, o.order_date, o.total
FROM users u
INNER JOIN orders o ON u.id = o.user_id
WHERE u.status = ‘active’
LIMIT 1000;
3. **使用适当的存储引擎**：
```sql
-- 为需要事务支持的表使用InnoDB
ALTER TABLE orders ENGINE = InnoDB;
-- 为只读或频繁读取的表使用MyISAM（如果不需要事务支持）
ALTER TABLE logs ENGINE = MyISAM;
配置优化
根据应用需求优化MySQL配置：
调整缓冲区大小
：
“`ini
my.cnf 或 my.ini 配置文件
[mysqld]
InnoDB缓冲池大小（建议设置为系统内存的50-70%）
innodb_buffer_pool_size = 4G
# 查询缓存大小（MySQL 8.0已移除此功能）
query_cache_size = 256M
query_cache_type = 1
2. **调整连接参数**：
```ini
[mysqld]
# 最大连接数
max_connections = 200
# 连接超时时间
wait_timeout = 300
interactive_timeout = 300
日志配置
：
“`ini
[mysqld]
错误日志
log_error = /var/log/mysql/error.log
# 慢查询日志
slow_query_log = 1
slow_query_log_file = /var/log/mysql/slow.log
long_query_time = 1
## 常见问题和解决方案
### 数据类型不兼容问题
**问题**：SQLite的某些数据类型在MySQL中没有直接对应。
**解决方案**：
1. 使用适当的数据类型映射：
```sql
-- SQLite中的BLOB类型
CREATE TABLE files (
id INTEGER PRIMARY KEY,
data BLOB
);
-- MySQL中的对应类型
CREATE TABLE files (
id INT AUTO_INCREMENT PRIMARY KEY,
data LONGBLOB
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
对于特殊数据类型，创建转换函数：
“`python
def convert_sqlite_boolean_to_mysql(value):
“”“将SQLite的布尔值转换为MySQL的TINYINT(1)”“”
if value is None:
return None
return 1 if value else 0
def convert_sqlite_datetime_to_mysql(value):
"""将SQLite的文本日期时间转换为MySQL的DATETIME"""
if value is None:
return None
# SQLite可能存储为各种格式，需要根据实际情况调整
try:
# 尝试解析ISO格式
return datetime.datetime.fromisoformat(value)
except:
# 尝试其他格式
return datetime.datetime.strptime(value, "%Y-%m-%d %H:%M:%S")
### 自定义函数和触发器迁移
**问题**：SQLite中的自定义函数和触发器需要重写以适应MySQL。
**解决方案**：
1. 将SQLite触发器转换为MySQL触发器：
```sql
-- SQLite触发器
CREATE TRIGGER update_user_modified
AFTER UPDATE ON users
FOR EACH ROW
BEGIN
UPDATE users SET modified = datetime('now') WHERE id = NEW.id;
END;
-- MySQL触发器
DELIMITER //
CREATE TRIGGER update_user_modified
AFTER UPDATE ON users
FOR EACH ROW
BEGIN
UPDATE users SET modified = NOW() WHERE id = NEW.id;
END; //
DELIMITER ;
使用存储过程替代SQLite的自定义函数：
“`sql
– MySQL存储过程示例
DELIMITER //
CREATE PROCEDURE calculate_order_total(IN order_id INT, OUT total DECIMAL(10,2))
BEGIN
SELECT SUM(price * quantity) INTO total
FROM order_items
WHERE order_id = order_id;
END; //
DELIMITER ;
– 调用存储过程
CALL calculate_order_total(123, @total);
SELECT @total;
### 外键约束问题
**问题**：SQLite默认不启用外键约束，而MySQL（InnoDB）默认启用。
**解决方案**：
1. 迁移前检查SQLite中的外键约束：
```sql
-- 启用SQLite外键约束
PRAGMA foreign_keys = ON;
-- 检查外键约束是否会导致问题
PRAGMA foreign_key_check;
在MySQL中正确设置外键约束：
“`sql
– 创建表时定义外键
CREATE TABLE order_items (
id INT AUTO_INCREMENT PRIMARY KEY,
order_id INT,
product_id INT,
quantity INT,
price DECIMAL(10,2),
FOREIGN KEY (order_id) REFERENCES orders(id) ON DELETE CASCADE,
FOREIGN KEY (product_id) REFERENCES products(id) ON DELETE RESTRICT
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
– 或者后期添加外键
ALTER TABLE order_items
ADD CONSTRAINT fk_order
FOREIGN KEY (order_id) REFERENCES orders(id) ON DELETE CASCADE;
### 性能问题
**问题**：迁移后查询性能不如预期。
**解决方案**：
1. 优化索引策略：
```sql
-- 添加适当的索引
CREATE INDEX idx_orders_customer_date ON orders(customer_id, order_date);
-- 使用覆盖索引
CREATE INDEX idx_order_items_order_product ON order_items(order_id, product_id, price);
优化查询语句：
“`sql
– 避免使用SELECT *
SELECT id, name, email FROM users WHERE status = ‘active’;
– 使用JOIN替代子查询
SELECT u.id, u.name, COUNT(o.id) as order_count
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
GROUP BY u.id, u.name;
3. 使用查询缓存：
```sql
-- MySQL 8.0之前的版本可以使用查询缓存
SELECT SQL_CACHE id, name FROM users WHERE status = 'active';
字符集和排序规则问题
问题
：SQLite和MySQL在字符集处理上有所不同。
解决方案
：
使用正确的字符集创建数据库和表：
“`sql
– 创建数据库时指定字符集
CREATE DATABASE my_database CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
– 创建表时指定字符集
CREATE TABLE users (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci,
email VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE utf8mb4_unicode_ci;
2. 转换现有数据的字符集：
```sql
-- 转换表的字符集
ALTER TABLE users CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
-- 转换特定列的字符集
ALTER TABLE users MODIFY name VARCHAR(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
自动化迁移脚本
为了简化迁移过程，可以创建一个自动化脚本：
import sqlite3
import mysql.connector
import argparse
import sys
import re
import os
from datetime import datetime
class SQLiteToMySQLMigrator:
def __init__(self, sqlite_db_path, mysql_config, output_dir=None):
self.sqlite_db_path = sqlite_db_path
self.mysql_config = mysql_config
self.output_dir = output_dir or 'migration_output'
self.ensure_output_dir()
# 连接到SQLite
self.sqlite_conn = sqlite3.connect(sqlite_db_path)
self.sqlite_cursor = self.sqlite_conn.cursor()
# 连接到MySQL
self.mysql_conn = mysql.connector.connect(**mysql_config)
self.mysql_cursor = self.mysql_conn.cursor()
def ensure_output_dir(self):
"""确保输出目录存在"""
if not os.path.exists(self.output_dir):
os.makedirs(self.output_dir)
def get_sqlite_tables(self):
"""获取SQLite中的所有表"""
self.sqlite_cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
return [table[0] for table in self.sqlite_cursor.fetchall()]
def get_table_schema(self, table_name):
"""获取SQLite表的创建语句"""
self.sqlite_cursor.execute(f"SELECT sql FROM sqlite_master WHERE type='table' AND name='{table_name}';")
result = self.sqlite_cursor.fetchone()
return result[0] if result else None
def convert_schema_to_mysql(self, schema_sql):
"""将SQLite表结构转换为MySQL表结构"""
# 替换数据类型
schema_sql = re.sub(r'\bINTEGER PRIMARY KEY\b', 'INT AUTO_INCREMENT PRIMARY KEY', schema_sql)
schema_sql = re.sub(r'\bTEXT\b', 'VARCHAR(255)', schema_sql)
schema_sql = re.sub(r'\bREAL\b', 'DOUBLE', schema_sql)
schema_sql = re.sub(r'\bBOOLEAN\b', 'TINYINT(1)', schema_sql)
schema_sql = re.sub(r'\bBLOB\b', 'LONGBLOB', schema_sql)
# 添加MySQL引擎和字符集
schema_sql = schema_sql.rstrip(';') + ' ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;'
return schema_sql
def export_table_data(self, table_name):
"""将SQLite表数据导出为CSV文件"""
output_file = os.path.join(self.output_dir, f"{table_name}.csv")
self.sqlite_cursor.execute(f"SELECT * FROM {table_name}")
rows = self.sqlite_cursor.fetchall()
# 获取列名
column_names = [description[0] for description in self.sqlite_cursor.description]
with open(output_file, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(column_names) # 写入列名
writer.writerows(rows) # 写入数据
return output_file
def create_mysql_table(self, table_name, schema_sql):
"""在MySQL中创建表"""
try:
self.mysql_cursor.execute(schema_sql)
self.mysql_conn.commit()
print(f"成功在MySQL中创建表: {table_name}")
return True
except mysql.connector.Error as err:
print(f"创建表 {table_name} 时出错: {err}")
return False
def import_csv_to_mysql(self, csv_file, table_name):
"""将CSV数据导入到MySQL表"""
with open(csv_file, 'r', encoding='utf-8') as f:
reader = csv.reader(f)
header = next(reader) # 读取列名
# 构建INSERT语句
placeholders = ', '.join(['%s'] * len(header))
columns = ', '.join(header)
sql = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
# 批量插入数据
batch_size = 1000
batch = []
inserted_count = 0
for row in reader:
batch.append(row)
if len(batch) >= batch_size:
self.mysql_cursor.executemany(sql, batch)
self.mysql_conn.commit()
inserted_count += len(batch)
batch = []
print(f”已插入 {inserted_count} 条记录到 {table_name}”)
# 插入剩余数据
if batch:
self.mysql_cursor.executemany(sql, batch)
self.mysql_conn.commit()
inserted_count += len(batch)
print(f”完成导入 {inserted_count} 条记录到 {table_name}”)
return inserted_count
def verify_migration(self, table_name):
“””验证迁移结果”””
# 获取SQLite中的记录数
self.sqlite_cursor.execute(f”SELECT COUNT(*) FROM {table_name}”)
sqlite_count = self.sqlite_cursor.fetchone()[0]
# 获取MySQL中的记录数
self.mysql_cursor.execute(f”SELECT COUNT(*) FROM {table_name}”)
mysql_count = self.mysql_cursor.fetchone()[0]
if sqlite_count == mysql_count:
print(f”表 {table_name} 验证成功: 记录数一致 ({sqlite_count})”)
return True
else:
print(f”表 {table_name} 验证失败: 记录数不一致 (SQLite: {sqlite_count}, MySQL: {mysql_count})”)
return False
def migrate_table(self, table_name):
“””迁移单个表”””
print(f”开始迁移表: {table_name}”)
# 获取并转换表结构
schema_sql = self.get_table_schema(table_name)
if not schema_sql:
print(f”无法获取表 {table_name} 的结构”)
return False
mysql_schema = self.convert_schema_to_mysql(schema_sql)
# 在MySQL中创建表
if not self.create_mysql_table(table_name, mysql_schema):
return False
# 导出数据
csv_file = self.export_table_data(table_name)
# 导入数据到MySQL
self.import_csv_to_mysql(csv_file, table_name)
# 验证迁移结果
return self.verify_migration(table_name)
def migrate_all_tables(self):
“””迁移所有表”””
tables = self.get_sqlite_tables()
print(f”找到 {len(tables)} 个表需要迁移”)
results = {}
for table_name in tables:
results[table_name] = self.migrate_table(table_name)
# 输出迁移结果摘要
print(“\n迁移结果摘要:”)
success_count = sum(1 for result in results.values() if result)
print(f”成功迁移: {success_count}/{len(tables)} 个表”)
for table_name, success in results.items():
status = “成功” if success else “失败”
print(f” {table_name}: {status}”)
return success_count == len(tables)
def close(self):
“””关闭数据库连接”””
self.sqlite_conn.close()
self.mysql_conn.close()
def main():
parser = argparse.ArgumentParser(description=’从SQLite迁移到MySQL’)
parser.add_argument(‘sqlite_db’, help=’SQLite数据库文件路径’)
parser.add_argument(‘–mysql-host’, default=’localhost’, help=’MySQL主机地址’)
parser.add_argument(‘–mysql-user’, required=True, help=’MySQL用户名’)
parser.add_argument(‘–mysql-password’, required=True, help=’MySQL密码’)
parser.add_argument(‘–mysql-database’, required=True, help=’MySQL数据库名’)
parser.add_argument(‘–output-dir’, default=’migration_output’, help=’输出目录’)
args = parser.parse_args()
mysql_config = {
‘host’: args.mysql_host,
‘user’: args.mysql_user,
‘password’: args.mysql_password,
‘database’: args.mysql_database
}
migrator = SQLiteToMySQLMigrator(args.sqlite_db, mysql_config, args.output_dir)
try:
success = migrator.migrate_all_tables()
sys.exit(0 if success else 1)
finally:
migrator.close()
if __name__ == ‘__main__’:
import csv
main()
使用这个自动化脚本：
python migrate.py database.sqlite \
–mysql-user your_username \
–mysql-password your_password \
–mysql-database your_database \
–output-dir migration_output
结论
从SQLite迁移到MySQL是一个复杂但可行的过程，需要仔细规划和执行。本文详细介绍了迁移的各个阶段，包括准备工作、结构迁移、数据迁移、验证和优化等关键步骤。通过使用适当的工具和方法，可以确保迁移过程的顺利进行，并在迁移后获得更好的性能和功能。
关键要点总结：
充分准备
：在开始迁移前，全面评估现有数据库结构、数据量和依赖关系。
结构转换
：正确处理SQLite和MySQL之间的数据类型和约束差异。
数据迁移
：选择适合数据量的迁移方法，确保数据完整性和一致性。
验证和优化
：迁移后进行全面的验证，并根据应用需求优化数据库性能。
问题处理
：提前了解可能遇到的问题，并准备相应的解决方案。
通过遵循本文提供的指南和方法，您可以成功地将数据库从SQLite迁移到MySQL，为应用的进一步发展奠定坚实的基础。