MongoDB数组更新最佳实践：DocumentDB批量更新内存问题优化求助

问题背景 #

我有一个包含数据数组的文档，需要更新数组中的多条数据，目前用array_filters实现，但查询效果差，更新3个月（约300条记录）的数据时触发内存不足错误：

Operation terminated due to low available memory

当前代码如下：

from pymongo import UpdateOne


hotel_id = 1
records = [
    {
        "date": "2021-01-01",
        "room_id": "1",
        "rate_id": "1",
        "price": 100,
        "ref_id": "",
    },
    {
        "date": "2021-01-02",
        "room_id": "1",
        "rate_id": "1",
        "price": 100,
        "ref_id": "",
    },
    {
        "date": "2021-01-03",
        "room_id": "1",
        "rate_id": "1",
        "price": 100,
        "ref_id": "",
    },
    {
        "date": "2021-01-01",
        "room_id": "2",
        "rate_id": "2",
        "price": 100,
        "ref_id": "",
    },
    {
        "date": "2021-01-02",
        "room_id": "2",
        "rate_id": "2",
        "price": 100,
        "ref_id": "",
    },
    {
        "date": "2021-01-03",
        "room_id": "2",
        "rate_id": "2",
        "price": 100,
        "ref_id": "",
    },
    {
        "date": "2021-01-01",
        "room_id": "3",
        "rate_id": "3",
        "price": 100,
        "ref_id": "",
    },
    {
        "date": "2021-01-02",
        "room_id": "3",
        "rate_id": "3",
        "price": 100,
        "ref_id": "",
    },
    {
        "date": "2021-01-03",
        "room_id": "3",
        "rate_id": "3",
        "price": 100,
        "ref_id": "",
    },
]

coll.bulk_write(
    [
        UpdateOne(
            filter={"hotel_id": hotel_id},
            update={
                "$set": {
                    f"data.$[i{index}].{key}": value
                    for index, record in enumerate(records)
                    for key, value in record.items()
                },
            },
            array_filters=[
                {
                    f"i{index}.date": record["date"],
                    f"i{index}.room_id": record["room_id"],
                    f"i{index}.rate_id": record["rate_id"],
                }
                for index, record in enumerate(records)
            ],
        )
    ]
)

优化方案 #

1. 拆分批量任务，控制单条更新复杂度 #

原代码把300条更新逻辑塞进单个UpdateOne，生成的array_filters和$set条目过多，直接触发内存过载。拆分多个小批量更新，每个批次处理20-50条数据（可根据DocumentDB内存限制调整）：

from pymongo import UpdateOne

hotel_id = 1
batch_size = 20  # 调整批次大小适配内存
update_tasks = []

# 按批次拆分记录
for i in range(0, len(records), batch_size):
    batch = records[i:i+batch_size]
    update_ops = {}
    filter_list = []
    for idx, record in enumerate(batch):
        # 构建当前元素的更新字段
        for key, value in record.items():
            update_ops[f"data.$[f{idx}].{key}"] = value
        # 构建对应的数组过滤条件
        filter_list.append({
            f"f{idx}.date": record["date"],
            f"f{idx}.room_id": record["room_id"],
            f"f{idx}.rate_id": record["rate_id"]
        })
    # 添加批次任务
    update_tasks.append(
        UpdateOne(
            filter={"hotel_id": hotel_id},
            update={"$set": update_ops},
            array_filters=filter_list
        )
    )

# 执行批量更新
coll.bulk_write(update_tasks)

2. 按唯一键分组，减少重复过滤逻辑 #

观察数据结构，每个数组元素可通过date+room_id+rate_id唯一标识，按room_id+rate_id分组后，同组内的更新可复用部分过滤条件，减少array_filters的数量：

from pymongo import UpdateOne
from collections import defaultdict

hotel_id = 1

# 按room_id和rate_id分组记录
grouped = defaultdict(list)
for record in records:
    key = (record["room_id"], record["rate_id"])
    grouped[key].append(record)

update_tasks = []
for (room_id, rate_id), group in grouped.items():
    update_ops = {}
    filter_list = []
    for idx, record in enumerate(group):
        update_ops[f"data.$[f{idx}].price"] = record["price"]
        update_ops[f"data.$[f{idx}].ref_id"] = record["ref_id"]
        filter_list.append({
            f"f{idx}.date": record["date"],
            f"f{idx}.room_id": room_id,
            f"f{idx}.rate_id": rate_id
        })
    update_tasks.append(
        UpdateOne(
            filter={"hotel_id": hotel_id},
            update={"$set": update_ops},
            array_filters=filter_list
        )
    )

coll.bulk_write(update_tasks)

3. 为数组查询字段创建复合索引 #

如果数组元素的查询字段（date、room_id、rate_id）无索引，DocumentDB会执行全数组扫描，增加内存消耗。创建复合索引加速定位：

// 在DocumentDB控制台或MongoDB Shell执行
db.collection.createIndex({ "hotel_id": 1, "data.date": 1, "data.room_id": 1, "data.rate_id": 1 })

4. 全量替换数组（适合大范围更新场景） #

如果需要更新数组中大部分元素，或者可以重新生成完整数组，直接替换整个数组的性能远高于逐条更新：

# 获取原文档的数组数据
original_doc = coll.find_one({"hotel_id": hotel_id}, {"data": 1})
if original_doc:
    # 用字典存储原数组元素，按唯一键去重覆盖
    data_map = {(item["date"], item["room_id"], item["rate_id"]): item for item in original_doc["data"]}
    for record in records:
        key = (record["date"], record["room_id"], record["rate_id"])
        # 合并新数据到原元素（覆盖更新）
        data_map[key] = {**data_map.get(key, {}), **record}
    # 生成新数组并替换
    new_data = list(data_map.values())
    coll.update_one({"hotel_id": hotel_id}, {"$set": {"data": new_data}})

优化核心逻辑 #

• 降低单条更新的复杂度：避免单个UpdateOne生成数百个过滤条件和更新字段，减少内存占用。
• 复用过滤条件：通过分组减少重复的过滤逻辑，提升查询匹配效率。
• 索引加速定位：复合索引让DocumentDB快速找到需要更新的数组元素，避免全数组扫描。
• 全量替换替代逐条更新：大范围更新场景下，全量替换的性能优势明显。

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