Pymongo无需遍历文档批量更新utc字段为10分钟取整时间的实现方法

可以实现，根据你使用的MongoDB版本和是否必须使用自定义Python函数，有两种最优实现方案：

MongoDB 4.2开始支持在update_many中传入聚合管道作为更新参数，直接在数据库服务端完成字段计算，不需要把全量数据拉取到本地，也不需要调用本地Python函数，全量更新1万条数据耗时通常在毫秒级。

你需要的10分钟向下取整逻辑可以直接用MongoDB内置日期运算符实现，代码如下：

from pymongo import MongoClient

client = MongoClient()
collection = client.your_db.your_collection

# 如果你存储的utc是ISO日期类型，用以下更新管道
update_pipeline = [
    {
        "$set": {
            "utc": {
                "$dateFromParts": {
                    "year": {"$year": "$utc"},
                    "month": {"$month": "$utc"},
                    "day": {"$dayOfMonth": "$utc"},
                    "hour": {"$hour": "$utc"},
                    "minute": {"$multiply": [{"$floor": {"$divide": [{"$minute": "$utc"}, 10]}}, 10]}
                }
            }
        }
    }
]

# 如果你存储的utc是秒级时间戳整数，用以下转换逻辑即可
# update_pipeline = [
#     {
#         "$set": {
#             "utc": {
#                 "$toLong": {
#                     "$dateFromParts": {
#                         "year": {"$year": {"$toDate": {"$multiply": ["$utc", 1000]}}},
#                         "month": {"$month": {"$toDate": {"$multiply": ["$utc", 1000]}}},
#                         "day": {"$dayOfMonth": {"$toDate": {"$multiply": ["$utc", 1000]}}},
#                         "hour": {"$hour": {"$toDate": {"$multiply": ["$utc", 1000]}}},
#                         "minute": {"$multiply": [{"$floor": {"$divide": [{"$minute": {"$toDate": {"$multiply": ["$utc", 1000]}}, 10]}}, 10]}
#                     }
#                 }
#             } / 1000  # 转回秒级时间戳
#         }
#     }
# ]

# 执行全量更新
collection.update_many({}, update_pipeline)

如果你的MongoDB版本低于4.2，或者必须依赖你自己写的round_to_10min函数处理逻辑，可以使用pymongo的bulk_write接口批量提交更新请求，相比逐条调用update_one性能提升至少一个数量级。

代码实现如下：

from pymongo import MongoClient
from pymongo.operations import UpdateOne
from bson import ObjectId
import utility_func

client = MongoClient()
collection = client.your_db.your_collection

# 只拉取需要的_id和utc字段，减少数据传输开销
alldocs = collection.find({}, {"_id": 1, "utc": 1})
update_ops = []

for x in alldocs:
    old_value = int(x['utc'])
    new_value = utility_func.round_to_10min(old_value)
    # 把更新操作加入批量列表
    update_ops.append(
        UpdateOne(
            {"_id": ObjectId(x['_id'])},
            {"$set": {"utc": new_value}}
        )
    )

# 一次提交所有更新操作
if update_ops:
    collection.bulk_write(update_ops)

• 操作前请先备份集合数据，避免更新逻辑出错导致数据丢失
• 如果后续集合数据量增长到10万条以上，建议增加分页查询逻辑分批处理，避免单次操作占用过多数据库资源

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