如何使用Python将MySQL数据直接迁移为MongoDB嵌入式文档

核心实现思路 #

你之前尝试的MySQL JOIN建视图的方案走不通，本质是JOIN返回的是扁平化的笛卡尔积结果，每一条子表数据都会重复对应一行父表数据，没法直接生成嵌套数组结构，完全没必要在MySQL层做嵌套转换，在Python层做分组聚合逻辑更简单、性能也更好，核心流程遵循以下原则：

• 禁止逐行做JOIN查询，这种方式会产生大量重复的父表数据，网络IO和内存浪费极其严重
• 先读取子表（address）全量数据，按关联外键user_id分组聚合，生成「用户ID-地址列表」的映射字典
• 分批读取父表（user）数据，给每个用户文档挂载对应分组好的地址子文档数组，凑够批次大小就批量写入MongoDB
• 大表场景必须做分批处理，禁止一次性把全量表数据加载到内存，避免进程OOM

参考实现代码 #

先安装依赖包：
pip install pymongo mysql-connector-python

可直接改造使用的代码：

import mysql.connector
from pymongo import MongoClient
from collections import defaultdict

# 连接配置，按需修改
MYSQL_CONF = {
    "host": "127.0.0.1",
    "port": 3306,
    "user": "your_mysql_account",
    "password": "your_mysql_password",
    "database": "USER"
}
MONGO_URI = "mongodb://127.0.0.1:27017/"
BATCH_SIZE = 1000  # 单批写入条数，根据机器内存调整

def run_migration():
    # 初始化数据库连接
    mysql_conn = mysql.connector.connect(**MYSQL_CONF)
    # 配置cursor返回字典格式结果，方便直接处理
    mysql_cursor = mysql_conn.cursor(dictionary=True)
    mongo_collection = MongoClient(MONGO_URI)["user_db"]["user"]

    # 第一步：拉取全量地址数据，按user_id分组
    address_group = defaultdict(list)
    mysql_cursor.execute("SELECT user_id, address_line_1, address_line_2, city, postal_code FROM address")
    for addr in mysql_cursor:
        uid = addr.pop("user_id") # 移除关联外键，不需要存入子文档
        address_group[uid].append(addr)

    # 第二步：分批拉取用户数据，挂载地址后批量写入MongoDB
    mysql_cursor.execute("SELECT id AS _id, name FROM user")
    write_batch = []
    for user in mysql_cursor:
        user["addresses"] = address_group.get(user["_id"], [])
        write_batch.append(user)
        # 达到批次大小触发批量写入
        if len(write_batch) >= BATCH_SIZE:
            mongo_collection.insert_many(write_batch, ordered=False)
            write_batch.clear()
    # 写入最后一批剩余数据
    if write_batch:
        mongo_collection.insert_many(write_batch, ordered=False)

    # 释放连接资源
    mysql_cursor.close()
    mysql_conn.close()

if __name__ == "__main__":
    run_migration()

性能优化参考 #

• 百万级以内数据量：上述基础代码足够用，子表分组的字典内存开销极低，100w条地址记录的内存占用通常在200M以内，普通开发机几分钟就能跑完迁移
• 千万级以上大表场景：不要一次性加载全量子表数据，先给子表关联外键user_id、父表主键id加上索引，按ID范围分段扫描父表和子表，用归并方式匹配挂载子文档，内存占用可以控制在固定值，和总数据量无关
• MongoDB写入侧必须用insert_many做批量写入，不要单条循环插入，批量写入性能是单条写入的几十到上百倍；加上ordered=False参数可以关闭MongoDB的写入顺序校验，还能再提升30%左右的写入速度
• 迁移前可以临时删掉目标集合的二级索引、将写入确认级别调整为w:1，等全量数据迁移完成后再重建索引，能减少索引维护带来的写入开销，速度提升明显
• 如果是多层嵌套结构（比如用户→订单→商品），逻辑完全通用：从最内层的子表开始逐层按外键分组，向上挂载到上层文档即可，不需要写复杂的多表JOIN SQL

避坑提示：如果子表存在软删除标记、数据过滤条件，记得在查询子表时加上对应的WHERE条件，避免把无效数据迁移到MongoDB中。

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