如何使用Python/Django将MySQL旧库数据迁移至已有表结构的新库

以下两种方案均兼容旧表缺少新表字段的场景，核心逻辑为插入时仅取新旧表共有的字段，新表新增字段自动沿用建表时设置的默认值（需提前确认新增字段允许为NULL或已配置默认值，无默认值且非空的字段需在迁移脚本中手动指定插入值）。

方案1：纯Python实现（无Django依赖） #

适合无Django运行环境的场景，依赖轻量、部署灵活。

• 先安装依赖包：
  pip install pymysql pandas sqlalchemy
• 迁移代码示例：

import pymysql
import pandas as pd
from sqlalchemy import create_engine

# 旧数据库连接配置
OLD_DB = {
    "host": "旧库IP/域名",
    "user": "旧库用户名",
    "password": "旧库密码",
    "db": "旧库名",
    "port": 3306,
    "charset": "utf8mb4"
}
# 新数据库连接配置
NEW_DB = {
    "host": "新库IP/域名",
    "user": "新库用户名",
    "password": "新库密码",
    "db": "新库名",
    "port": 3306,
    "charset": "utf8mb4"
}

# 初始化数据库连接引擎
old_engine = create_engine(f"mysql+pymysql://{OLD_DB['user']}:{OLD_DB['password']}@{OLD_DB['host']}:{OLD_DB['port']}/{OLD_DB['db']}?charset={OLD_DB['charset']}")
new_engine = create_engine(f"mysql+pymysql://{NEW_DB['user']}:{NEW_DB['password']}@{NEW_DB['host']}:{NEW_DB['port']}/{NEW_DB['db']}?charset={NEW_DB['charset']}")

# 读取旧库所有表名
with old_engine.connect() as conn:
    table_list = pd.read_sql("SHOW TABLES", conn).iloc[:, 0].tolist()

# 逐表迁移
for table in table_list:
    print(f"开始迁移表：{table}")
    # 分批读取避免大表内存溢出，单批大小可根据服务器配置调整
    for df_chunk in pd.read_sql(f"SELECT * FROM {table}", old_engine, chunksize=1000):
        # 读取新表字段列表
        new_table_cols = pd.read_sql(f"DESC {table}", new_engine)["Field"].tolist()
        # 仅保留新旧表共有的字段
        common_cols = [col for col in df_chunk.columns if col in new_table_cols]
        df_chunk = df_chunk[common_cols]
        # 追加写入新表
        df_chunk.to_sql(
            name=table,
            con=new_engine,
            if_exists="append",
            index=False
        )
    print(f"表{table}迁移完成")

方案2：Django环境实现 #

适合已使用Django对接新旧数据库的场景，可直接复用Django的数据库连接配置。

• 第一步：在settings.py中配置双数据源

DATABASES = {
    # 新数据库（默认连接）
    "default": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "新库名",
        "USER": "新库用户名",
        "PASSWORD": "新库密码",
        "HOST": "新库IP/域名",
        "PORT": 3306,
        "CHARSET": "utf8mb4"
    },
    # 旧数据库
    "old_db": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "旧库名",
        "USER": "旧库用户名",
        "PASSWORD": "旧库密码",
        "HOST": "旧库IP/域名",
        "PORT": 3306,
        "CHARSET": "utf8mb4"
    }
}

• 第二步：编写迁移代码（可封装为自定义Django命令执行）

from django.db import connections

# 初始化数据库游标
old_cursor = connections["old_db"].cursor()
new_cursor = connections["default"].cursor()

# 获取旧库所有表名
old_cursor.execute("SHOW TABLES")
table_list = [item[0] for item in old_cursor.fetchall()]

BATCH_SIZE = 1000 # 单批插入条数，可调整

for table in table_list:
    print(f"开始迁移表：{table}")
    # 获取旧表字段
    old_cursor.execute(f"DESC {table}")
    old_cols = [item[0] for item in old_cursor.fetchall()]
    # 获取新表字段
    new_cursor.execute(f"DESC {table}")
    new_cols = [item[0] for item in new_cursor.fetchall()]
    # 取公共字段
    common_cols = [col for col in old_cols if col in new_cols]
    col_str = ",".join(common_cols)
    placeholder_str = ",".join(["%s"] * len(common_cols))
    
    offset = 0
    while True:
        # 分批读取旧表数据
        old_cursor.execute(
            f"SELECT {col_str} FROM {table} LIMIT %s OFFSET %s",
            [BATCH_SIZE, offset]
        )
        batch_data = old_cursor.fetchall()
        if not batch_data:
            break
        # 批量写入新表
        new_cursor.executemany(
            f"INSERT INTO {table} ({col_str}) VALUES ({placeholder_str})",
            batch_data
        )
        offset += BATCH_SIZE
    print(f"表{table}迁移完成")

迁移注意事项 #

• 迁移前务必备份新旧库全量数据，避免操作失误导致数据丢失
• 大表迁移时可适当调整单批处理大小，平衡内存占用和迁移速度
• 迁移完成后需做数据一致性校验：对比同表行数、关键字段的聚合结果（求和、计数、去重计数）是否一致
• 若新库新增字段为非空且无默认值，需在插入前手动给该字段赋值，例如在公共字段后新增该字段，统一填充默认值

内容的提问来源于stack exchange，提问作者arian_radmehr