MongoDB通过MongoEngine批量插入更新CSV数据的最优方案

原实现的核心性能问题是每条记录需要执行1次查询+1次写入，共2次网络IO，400万条记录至少产生800万次IO请求，叠加MongoEngine单条save方法的封装开销，整体执行效率极低。你已经为company_code和date建立了唯一复合索引，完全可以借助MongoDB原生的批量upsert能力实现性能跃升，无需提前逐个查询记录是否存在。

方案1：基于MongoEngine的bulk_write实现 #

无需修改现有CSV解析逻辑，直接将对象列表转换为批量操作请求提交即可，性能可提升数十倍：

from pymongo import UpdateOne

# 按批次处理避免内存占用过高，可根据服务器内存调整为5000~20000
BATCH_SIZE = 10000
operations = []

for inf in your_daily_report_list:
    operations.append(
        UpdateOne(
            # 匹配业务主键
            filter={"company_code": inf.company_code, "date": inf.date},
            # 全字段更新
            update={"$set": inf.to_mongo().to_dict()},
            # 匹配不到则插入新记录
            upsert=True
        )
    )
    # 批次满了就提交
    if len(operations) >= BATCH_SIZE:
        # ordered=False允许MongoDB并行执行操作，速度更快
        DailyReport.objects.bulk_write(operations, ordered=False)
        operations = []
# 提交剩余的操作
if operations:
    DailyReport.objects.bulk_write(operations, ordered=False)

方案2：极致性能优化（跳过MongoEngine对象层） #

如果不需要用到MongoEngine的字段验证、信号触发等能力，直接调用底层pymongo集合接口，同时解析CSV时直接构造字典，省去MongoEngine对象实例化的开销，速度可再提升30%以上：

from pymongo import UpdateOne
import csv
from datetime import datetime

BATCH_SIZE = 10000
# 获取原生pymongo集合对象
coll = DailyReport._get_collection()
operations = []

with open("你的CSV文件路径", "r") as f:
    reader = csv.DictReader(f)
    for row in reader:
        # 直接把CSV行转换为对应数据格式，无需实例化DailyReport
        doc = {
            "company_code": row["company_code"],
            "date": datetime.strptime(row["date"], "你的日期格式"),
            # 其余字段按需求转换，示例：
            "revenue": float(row["revenue"]),
            "employee_count": int(row["employee_count"])
            # ... 共8个字段
        }
        operations.append(
            UpdateOne(
                filter={"company_code": doc["company_code"], "date": doc["date"]},
                update={"$set": doc},
                upsert=True
            )
        )
        if len(operations) >= BATCH_SIZE:
            coll.bulk_write(operations, ordered=False)
            operations = []
if operations:
    coll.bulk_write(operations, ordered=False)

额外优化建议 #

• 导入前如果集合有其他非必要索引，可以先临时删除，导入完成后再重建，能进一步提升写入速度
• CSV解析时如果有重复的company_code+date记录，提前在内存中去重，避免无效的更新操作

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