如何用MongoDB的UpdateMany替代UpdateOne实现批量数据同步

误区说明 #

UpdateMany 仅适用于给匹配同一筛选条件的多个文档应用相同更新操作的场景（比如批量修改某类状态的文档）。而你的需求是每个_id对应唯一文档，且每个文档的更新内容各不相同，因此无法用单个UpdateMany完成所有操作，但可以通过以下两种方式优化批量同步的效率：

方案1：优化原有批量操作（低改动、内存友好） #

原代码将所有UpdateOne请求存入列表会占用大量内存，改用生成器表达式可以逐个生成请求，避免内存溢出，同时保持逻辑一致：

from pymongo import MongoClient, UpdateOne
from datetime import datetime
import pandas as pd

client = MongoClient()
database = client['Eg']
collection = database['eg']
start = datetime.now()

df = pd.read_csv("eg.csv")
df['_id'] = df["Factory_Id"] + df["Order ID"]
data = df.to_dict(orient="records")

# 用生成器表达式替代列表，按需生成UpdateOne请求
requests = (UpdateOne({'_id': doc['_id']}, {'$set': doc}, upsert=True) for doc in data)

# 执行批量操作，ordered=False可提升大数量场景下的效率（允许并行处理）
result = collection.bulk_write(requests, ordered=False)

print(f"完成：更新{result.modified_count}条，插入{result.upserted_count}条，耗时{datetime.now()-start}")

优势：

• 代码改动极小，保留原有业务逻辑
• 生成器不占用额外内存存储所有请求，适合百万级数据
• ordered=False关闭有序执行，MongoDB可并行处理请求，提升速度

方案2：利用MongoDB聚合管道$merge（超大数据量首选） #

如果数据量极大（千万级以上），可以将数据先导入临时集合，再通过MongoDB的$merge聚合管道完成批量Upsert，所有处理在数据库端完成，大幅减少网络交互开销：

from pymongo import MongoClient
from datetime import datetime
import pandas as pd

client = MongoClient()
database = client['Eg']
target_col = database['eg']
temp_col = database['temp_eg']
start = datetime.now()

df = pd.read_csv("eg.csv")
df['_id'] = df["Factory_Id"] + df["Order ID"]

# 清空并导入临时集合
temp_col.drop()
temp_col.insert_many(df.to_dict(orient="records"))

# 通过$merge完成批量Upsert：匹配_id，存在则替换，不存在则插入
target_col.aggregate([
    {"$merge": {
        "into": "eg",
        "on": "_id",
        "whenMatched": "replace",  # 若仅需更新部分字段，可改用{"$set": {"字段1": "$$new.字段1", ...}}
        "whenNotMatched": "insert"
    }}
])

print(f"完成：耗时{datetime.now()-start}")

注意：

• 需要MongoDB 4.2及以上版本支持$merge
• 若仅需更新部分字段，可将whenMatched改为自定义更新逻辑，比如：

  "whenMatched": {"$set": {"Amount": "$$new.Amount", "Status": "$$new.Status"}}
  

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