如何用PyMongo高效实现MongoDB文档的批量Upsert？

要实现「存在指定id_number则更新（仅数据变更时），不存在则插入」的需求，最高效的方式是批量操作+索引优化，具体步骤如下：

1. 先给id_number建唯一索引 #

这是核心前提，没有索引的话，9万条数据的匹配查询会慢到无法接受。执行以下代码创建索引：

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["你的数据库名"]
collection = db["你的集合名"]

# 创建唯一索引，确保id_number唯一，同时加速查询
collection.create_index("id_number", unique=True)

2. 用bulk_write配合UpdateOne实现批量UPSERT #

单条循环操作会产生大量网络请求，效率极低，bulk_write可以一次性提交一批操作，大幅减少交互次数。

完整代码示例 #

from pymongo import MongoClient, UpdateOne

# 初始化连接
client = MongoClient("mongodb://localhost:27017/")
db = client["你的数据库名"]
collection = db["你的集合名"]

# 你的原始数据
data = [
  {"name": "John", "id_number":1, "age": "20"},
  {"name": "Bob", "id_number":2, "age": 19}
]

# 构造批量操作列表
operations = []
for item in data:
    # 匹配条件：找到对应id_number的文档
    filter_condition = {"id_number": item["id_number"]}
    # 更新规则：设置所有字段，MongoDB会自动判断内容是否变更，无变更则不执行更新
    update_action = {"$set": item}
    # 添加UPSERT操作：匹配到则更新，没匹配到则插入
    operations.append(UpdateOne(filter_condition, update_action, upsert=True))

# 分批次执行，避免内存压力过大（建议每1000-5000条一批）
batch_size = 1000
for i in range(0, len(operations), batch_size):
    batch_ops = operations[i:i+batch_size]
    result = collection.bulk_write(batch_ops)
    print(f"第{i//batch_size +1}批处理完成：插入{result.upserted_count}条，更新{result.modified_count}条")

为什么这个方案高效？ #

• 减少网络交互：单条循环需要向MongoDB发送9万次请求，分批次后仅需几十次，大幅降低网络开销
• 索引加速匹配：唯一索引让id_number的查询从全表扫描变成O(1)的快速定位
• MongoDB原生优化：bulk_write是MongoDB原生支持的批量操作，在服务器端批量执行，性能远高于客户端循环

注意事项 #

• 批次大小可根据服务器内存和网络调整，不要一次性提交9万条，避免内存溢出
• 如果数据中包含_id字段，要确保和id_number逻辑一致，避免冲突
• 先拿小批量数据测试逻辑，确认插入/更新符合预期后再处理全量数据

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