如何通过PyMongo批量清理MongoDB大集合中值为空字符串的字段

完全可以通过PyMongo实现该需求，针对亿级大文档量、多字段的场景，优先选择MongoDB 4.2+支持的服务端聚合管道更新方案，不需要把文档拉取到客户端、不需要提前枚举所有字段，性能远高于遍历文档后逐个$unset的方案。

利用聚合管道的$objectToArray+$filter+$arrayToObject组合，自动过滤所有值为空字符串的字段，直接在MongoDB服务端完成文档更新：

1. 用$objectToArray把当前文档（$$ROOT）转成[{k:字段名, v:字段值}, ...]格式的数组
2. 用$filter过滤掉数组中v等于空字符串的项
3. 用$arrayToObject把过滤后的数组转回文档，替换原文档即可自动移除所有空字符串字段

基础版（仅移除顶层空字符串字段） #

import pymongo
from pymongo import WriteConcern

# 初始化MongoDB连接，按需修改连接参数
client = pymongo.MongoClient("mongodb://你的连接地址")
db = client["你的库名"]
col = db["你的集合名"]

# 聚合更新管道
update_pipeline = [
    {
        "$replaceWith": {
            "$arrayToObject": {
                "$filter": {
                    "input": {"$objectToArray": "$$ROOT"},
                    "cond": {"$ne": ["$$this.v", ""]}
                }
            }
        }
    }
]

# 全集合更新，w=0的写策略关闭acknowledge提升写入速度，生产环境可按需调整
update_result = col.update_many(
    filter={},
    update=update_pipeline,
    write_concern=WriteConcern(w=0)
)

大集合优化版（分批执行避免锁表） #

针对2000万+、后续到亿级的集合，建议在业务低峰期分批执行，避免长时间占住写锁影响正常业务：

import pymongo

client = pymongo.MongoClient("mongodb://你的连接地址")
db = client["你的库名"]
col = db["你的集合名"]

update_pipeline = [
    {
        "$replaceWith": {
            "$arrayToObject": {
                "$filter": {
                    "input": {"$objectToArray": "$$ROOT"},
                    "cond": {"$ne": ["$$this.v", ""]}
                }
            }
        }
    }
]

BATCH_SIZE = 10000  # 每批处理1万条，可根据服务器性能调整
last_id = None

while True:
    # 构造分批查询条件
    query = {}
    if last_id:
        query["_id"] = {"$gt": last_id}
    
    # 拉取一批文档的_id
    batch = list(col.find(query, {"_id": 1}).limit(BATCH_SIZE))
    if not batch:
        break
    
    # 仅更新当前批次的文档
    batch_ids = [item["_id"] for item in batch]
    col.update_many({"_id": {"$in": batch_ids}}, update_pipeline)
    
    # 更新游标位置
    last_id = batch[-1]["_id"]

扩展版（支持移除嵌套文档内的空字符串字段） #

如果需要处理嵌套层级的空字符串字段，可以使用MongoDB 4.4+支持的$function自定义JS逻辑实现递归过滤：

update_pipeline = [
    {
        "$replaceWith": {
            "$function": {
                "body": """function(doc) {
                    function removeEmpty(obj) {
                        for (const key in obj) {
                            if (obj[key] === "") delete obj[key];
                            else if (typeof obj[key] === "object" && obj[key] !== null) removeEmpty(obj[key]);
                        }
                        return obj;
                    }
                    return removeEmpty(doc);
                }""",
                "args": ["$$ROOT"],
                "lang": "js"
            }
        }
    }
]

• 执行操作前务必对集合做全量备份，避免误操作导致数据丢失
• 先取少量测试文档验证逻辑符合预期后，再执行全量更新
• 若MongoDB版本低于4.2不支持聚合管道更新，才需要选择遍历文档枚举空字段后调用$unset的方案，性能会比服务端管道低30%以上，优先考虑升级数据库版本

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