如何在PyMongo中实现批量Upsert操作？

当然可以实现批量Upsert！你的思路完全正确——减少客户端与MongoDB服务器之间的往返次数，确实是解决这类场景性能瓶颈的核心手段，毕竟单条请求的网络开销累加起来会非常可观。

在PyMongo中，我们可以通过bulk_write()方法结合ReplaceOne操作来实现批量的插入/更新逻辑，具体步骤如下：

1. 导入必要的类 #

首先需要从pymongo导入ReplaceOne操作类，它和你之前用的replace_one逻辑完全对应，只是用于批量操作场景：

from pymongo import ReplaceOne
import datetime

2. 构建批量操作列表并分批执行 #

我们可以设置一个批次大小（比如你说的5000条），遍历数据时将每个Upsert操作加入列表，当列表达到批次大小或遍历结束时，一次性发送给服务器执行：

# 设置批次大小
BATCH_SIZE = 5000
operations = []

for rowindex, row in InsertionData.iterrows():
    record = row.to_dict()
    # 统一设置当日日期为0点的datetime
    record['date'] = datetime.datetime.combine(datetime.date.today(), datetime.time(0, 0))
    
    # 构建ReplaceOne操作：filter匹配条件，replacement是要插入/更新的文档，upsert=True开启插入逻辑
    operation = ReplaceOne(
        filter={'student': row.get('student'), 'date': record['date']},
        replacement=record,
        upsert=True
    )
    operations.append(operation)
    
    # 当操作列表达到批次大小时，执行批量操作并清空列表
    if len(operations) == BATCH_SIZE:
        collection.bulk_write(operations)
        operations = []

# 处理最后一批不足BATCH_SIZE的操作
if operations:
    collection.bulk_write(operations)

3. 额外优化与注意事项 #

• 批次大小调整：MongoDB默认有最大消息大小限制（通常是16MB），5000条是比较保守的数值，如果你的单条文档体积很小，可以适当调大；如果文档较大，则需要调小，避免超过限制。
• 无序执行：如果你的业务允许部分操作失败不影响其他操作，可以在bulk_write中添加ordered=False参数，这样MongoDB会并行执行操作，且某条操作失败不会中断整个批次：

  collection.bulk_write(operations, ordered=False)
  

• 错误处理：可以添加try-except块来捕获批量操作中的异常，比如：

  try:
      collection.bulk_write(operations)
  except Exception as e:
      print(f"批量操作失败：{e}")
      # 这里可以根据业务需求添加重试或日志记录逻辑
  

这种方式将原来的N次网络请求压缩成ceil(N/5000)次，能极大减少服务器间通信的开销，完美解决你当前的性能瓶颈。

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