使用pymongo将MySQL数据迁移至MongoDB时数据类型报错如何解决

pymongo 对BSON类型有原生映射规则，只需对mysql.connector返回的四类特殊值做对应转换即可解决写入报错问题，具体方案如下：

各类型转换规则 #

• None值：pymongo原生支持Python的None直接转换为BSON的null类型，若出现写入报错，大概率是伴随字段名包含.、$等MongoDB非法字符的连带问题，排除字段名问题后直接保留None即可正常写入。
• datetime.date类型：MySQL的DATE类型查询结果为Python datetime.date，pymongo仅支持datetime.datetime类型转BSON日期，只需补全时分秒即可转换：datetime.datetime.combine(date_val, datetime.datetime.min.time())，有时区需求可自行追加tzinfo参数。
• datetime.timedelta类型：MySQL的TIME类型查询结果为Python datetime.timedelta，BSON无原生时间差类型，可按需选择两种转换方案：
  • 存为数值：转换为总毫秒/秒数整数int(timedelta_val.total_seconds() * 1000)，适合后续需要做数值计算的场景
  • 存为字符串：直接转str(timedelta_val)存储，适合仅做展示不需要计算的场景
• datetime.datetime类型：pymongo原生支持Python datetime.datetime转BSON的UTCDateTime，若MySQL返回的是带时区的datetime，建议先转成UTC时间再写入避免时区偏差；无时区的 naive datetime会被pymongo默认识别为UTC时间，有自定义时区需求可手动补充tzinfo参数。

完整实现代码 #

单条数据转换函数 #

import datetime
from pymongo import MongoClient
import mysql.connector

def convert_mysql_row_to_mongo(row):
    mongo_doc = {}
    for k, v in row.items():
        if v is None:
            mongo_doc[k] = v
        elif isinstance(v, datetime.date) and not isinstance(v, datetime.datetime):
            # date转datetime
            mongo_doc[k] = datetime.datetime.combine(v, datetime.datetime.min.time())
        elif isinstance(v, datetime.timedelta):
            # 此处采用转毫秒的方案，可按需替换为转字符串逻辑
            mongo_doc[k] = int(v.total_seconds() * 1000)
        elif isinstance(v, datetime.datetime):
            # 有 timezone 转换需求可在此处补充逻辑
            mongo_doc[k] = v
        else:
            # 字符串、数值、布尔等类型直接保留
            mongo_doc[k] = v
    return mongo_doc

500万行批量迁移代码 #

采用分批拉取、批量写入的逻辑，避免内存溢出，性能更高：

# MySQL 连接配置
mysql_conn = mysql.connector.connect(
    host="你的MySQL地址",
    user="你的MySQL账号",
    password="你的MySQL密码",
    database="你的库名",
    dictionary=True # 直接返回字典格式，无需手动映射字段名
)
mysql_cursor = mysql_conn.cursor(buffered=True)

# MongoDB 连接配置
mongo_client = MongoClient("mongodb://你的MongoDB地址:27017/")
mongo_db = mongo_client["目标库名"]
mongo_col = mongo_db["employees"]

# 每批拉取10000条，可根据服务器内存调整为1-2万
batch_size = 10000
offset = 0

while True:
    mysql_cursor.execute("SELECT * FROM employees LIMIT %s OFFSET %s", (batch_size, offset))
    rows = mysql_cursor.fetchall()
    if not rows:
        break
    # 批量转换
    doc_list = [convert_mysql_row_to_mongo(row) for row in rows]
    # 批量写入MongoDB
    mongo_col.insert_many(doc_list)
    offset += batch_size
    # 可选打印迁移进度
    print(f"已完成{offset}条数据迁移")

# 关闭连接
mysql_cursor.close()
mysql_conn.close()
mongo_client.close()

注意事项 #

• 如果需要将MySQL主键作为MongoDB的唯一键，可在转换函数中追加mongo_doc["_id"] = row["MySQL主键字段名"]，避免重复写入。
• 若TIME类型需要保留结构化信息，也可将timedelta转换为{"hours":x, "minutes":y, "seconds":z}的嵌套文档存储，根据业务查询需求选择即可。

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