PyMongo 3.6及以上版本跨数据库批量迁移文档的最优方案

我懂你现在的困扰——之前依赖的pymongo.bulk.BulkOperationBuilder已经被弃用，官方MongoDB的db.cloneCollection()在PyMongo里又找不到对应功能，copydb也不符合你的需求，还自己琢磨出了两个先批量插入再删原集合的方案。咱先聊聊你的这两个方案，再说说更高效的迁移思路。

先分析你提出的两个方案 #

• 方案1：用bulk_write配合InsertOne生成器，还加上了ordered=False。这个思路挺不错的——生成器不会一次性把所有文档加载到内存，适合处理大集合；ordered=False关闭顺序校验，还能提升批量写入的速度。唯一要注意的是，如果迁移过程中出现部分文档写入失败，ordered=False会继续处理剩下的文档，你需要事后检查result里的失败记录。
• 方案2：直接用insert_many+coll_from.find()。这个写法最简单，但有个致命隐患：coll_from.find()会把所有查询结果一次性拉到客户端内存里，如果集合数据量很大，很容易触发内存溢出，不推荐用于大集合迁移。

更优的服务器端迁移方案：聚合管道+$out/$merge #

其实MongoDB官方推荐的跨库批量迁移方式是服务器端聚合操作，也就是用$out或$merge阶段直接在数据库内部完成数据迁移，比客户端拉取再插入高效得多——毕竟省掉了客户端和服务器之间的大量网络传输，还不用客户端承担内存压力。

场景1：目标集合不存在，或需要完全覆盖目标集合 #

用$out阶段可以直接把聚合结果写入目标数据库的集合，若目标集合已存在，会被完全替换：

from pymongo import MongoClient

mongo = MongoClient()
coll_from = mongo['db_1']['coll_name']

# 用聚合管道的$out直接完成跨库写入
coll_from.aggregate([
    {"$out": {"db": "db_2", "coll": "coll_name"}}
])

# 确认迁移完成后删除原集合
mongo['db_1'].drop_collection('coll_name')

场景2：目标集合已存在，需要合并/追加文档 #

如果目标集合已经有数据，你想实现追加或去重合并，可以用$merge阶段，还能自定义冲突处理规则：

from pymongo import MongoClient

mongo = MongoClient()
coll_from = mongo['db_1']['coll_name']

coll_from.aggregate([
    {"$merge": {
        "into": {"db": "db_2", "coll": "coll_name"},
        "on": "_id",  # 以_id作为唯一匹配键，避免重复插入
        "whenMatched": "keepExisting",  # 遇到重复文档时，保留目标集合的现有数据
        "whenNotMatched": "insert"  # 没有匹配的文档直接插入
    }}
])

mongo['db_1'].drop_collection('coll_name')

注意事项 #

• 这个方案要求你的MongoDB版本在4.2及以上，因为从4.2开始$out和$merge才支持跨数据库操作。
• 迁移过程中尽量避免对原集合写入新数据，否则可能会有数据遗漏；如果必须写入，可以考虑先给原集合加读锁，或者用变更流同步增量数据。

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