PyMongo如何高效批量将文档中的空字符串替换为null

不要循环调用replace_one()做单条更新，这类场景直接使用MongoDB 4.2及以上版本支持的聚合管道更新能力，配合update_many()即可实现服务端批量执行，性能远高于客户端逐条操作。

场景1：已知需要处理的固定字段列表 #

这是最常用的场景，提前指定要检查替换的字段，通过$cond判断字段值为空字符串时替换为null，否则保留原值，同时通过过滤条件仅匹配存在空串的文档，避免无效扫描：

from pymongo import MongoClient
from bson.objectid import ObjectId

# 初始化连接
client = MongoClient("你的MongoDB连接地址")
collection = client["目标数据库名"]["目标集合名"]

# 配置需要检查替换的字段
target_fields = ["name", "phone", "email", "address", "remark"]

# 构造更新聚合管道
update_stage = {
    "$set": {
        field: {
            "$cond": {
                "if": {"$eq": [f"${field}", ""]},
                "then": None,
                "else": f"${field}"
            }
        } for field in target_fields
    }
}

# 构造过滤条件：仅匹配任意目标字段为空串的文档
filter_condition = {"$or": [{field: ""} for field in target_fields]}

# 执行批量更新
update_result = collection.update_many(filter_condition, [update_stage])
print(f"匹配文档数：{update_result.matched_count}，实际更新文档数：{update_result.modified_count}")

场景2：需要替换文档所有层级（含嵌套字段、数组）的空字符串 #

如果不确定字段名，需要无差别扫描所有字段、嵌套对象、数组元素，把其中的空字符串替换为null，可以在聚合管道中使用自定义JS函数递归处理（要求MongoDB 5.0及以上版本开启服务端脚本支持）：

# 递归替换空串的更新管道
full_replace_pipeline = [
    {
        "$replaceRoot": {
            "newRoot": {
                "$function": {
                    "body": """function processNode(node) {
                        if (node === "") return null;
                        if (Array.isArray(node)) return node.map(processNode);
                        if (typeof node === "object" && node !== null) {
                            Object.keys(node).forEach(key => {
                                node[key] = processNode(node[key]);
                            });
                        }
                        return node;
                    }""",
                    "args": ["$$ROOT"],
                    "lang": "js"
                }
            }
        }
    }
]

# 执行更新，空匹配条件会遍历所有文档，无空串的文档不会产生实际修改
collection.update_many({}, full_replace_pipeline)

大数据量优化建议 #

• 千万级以上文档集合不要一次性执行全量更新，建议按_id范围分批处理，单批处理1w-5w条即可，避免长事务锁影响线上业务：

batch_size = 20000
last_id = ObjectId("000000000000000000000000")
while True:
    batch_filter = {
        "_id": {"$gt": last_id},
        **filter_condition
    }
    batch_res = collection.update_many(batch_filter, [update_stage])
    if batch_res.matched_count == 0:
        break
    # 更新下一批次的起始_id
    last_doc = collection.find(batch_filter, {"_id": 1}).sort("_id", -1).limit(1).next()
    last_id = last_doc["_id"]

• 执行更新前务必先在测试环境验证逻辑，可以先取少量样本数据通过aggregate()执行同样的管道，确认替换结果符合预期再跑全量。
• 绝对不要把全量文档拉到客户端逐条修改再写回，这类操作的网络IO、序列化开销是服务端更新的上百倍，数据量稍大就会出现超时、内存占用过高的问题。

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