本文目录导读：
我来详细介绍几种批量更新MongoDB文档的方法和脚本示例。
使用 bulkWrite() 方法（推荐）
JavaScript/Node.js 示例
// 批量更新脚本
async function batchUpdateMongoDB() {
const MongoClient = require('mongodb').MongoClient;
// 连接数据库
const client = new MongoClient('mongodb://localhost:27017');
await client.connect();
const db = client.db('your_database');
const collection = db.collection('your_collection');
try {
// 准备批量更新操作
const bulkOps = [
{
updateOne: {
filter: { _id: ObjectId("...") },
update: { $set: { status: 'active', updatedAt: new Date() } }
}
},
{
updateOne: {
filter: { _id: ObjectId("...") },
update: { $set: { status: 'inactive' } }
}
}
];
// 执行批量更新
const result = await collection.bulkWrite(bulkOps);
console.log(`成功更新 ${result.modifiedCount} 条文档`);
console.log(`匹配 ${result.matchedCount} 条文档`);
} catch (error) {
console.error('批量更新失败:', error);
} finally {
await client.close();
}
}
基于条件的批量更新
Python 示例
from pymongo import MongoClient
from datetime import datetime
def batch_update_by_condition():
# 连接数据库
client = MongoClient('mongodb://localhost:27017')
db = client['your_database']
collection = db['your_collection']
# 方式1：使用 update_many
result = collection.update_many(
{"status": "pending", "created_at": {"$lt": datetime(2024, 1, 1)}},
{"$set": {"status": "expired", "updated_at": datetime.now()}}
)
print(f"更新了 {result.modified_count} 条文档")
# 方式2：批量处理不同条件
def batch_update_different_conditions():
operations = [
# 更新所有活跃用户
collection.update_many(
{"role": "user", "status": "active"},
{"$set": {"last_activity": datetime.now()}}
),
# 更新所有管理员
collection.update_many(
{"role": "admin"},
{"$inc": {"login_count": 1}}
)
]
for i, result in enumerate(operations):
print(f"操作{i+1}: 更新了 {result.modified_count} 条")
client.close()
# 执行批量更新
batch_update_by_condition()
逐条处理的大批量更新
Node.js 流式处理
async function batchStreamUpdate() {
const MongoClient = require('mongodb').MongoClient;
const client = new MongoClient('mongodb://localhost:27017');
await client.connect();
const db = client.db('your_database');
const collection = db.collection('your_collection');
const BATCH_SIZE = 1000; // 每批处理1000条
try {
// 使用游标批量处理
const cursor = collection.find({
// 你的查询条件
status: 'pending'
});
let batch = [];
let count = 0;
while (await cursor.hasNext()) {
const doc = await cursor.next();
// 处理文档
batch.push({
updateOne: {
filter: { _id: doc._id },
update: {
$set: {
processed: true,
processedAt: new Date()
}
}
}
});
count++;
// 达到批次大小则执行更新
if (batch.length >= BATCH_SIZE) {
const result = await collection.bulkWrite(batch);
console.log(`已处理 ${count} 条文档`);
batch = []; // 清空批次
}
}
// 处理最后一批
if (batch.length > 0) {
const result = await collection.bulkWrite(batch);
console.log(`最终处理了 ${count} 条文档`);
}
} catch (error) {
console.error('批量处理出错:', error);
} finally {
await client.close();
}
}
Shell 脚本批量更新
MongoDB Shell 脚本
// batch_update.js
var db = connect('mongodb://localhost:27017/your_database');
var collection = db.getCollection('your_collection');
// 批量更新条件
var batchSize = 1000;
var totalUpdated = 0;
var skip = 0;
while (true) {
var docs = collection.find({
status: 'pending',
category: 'old'
}).skip(skip).limit(batchSize).toArray();
if (docs.length === 0) {
break;
}
var bulkOps = docs.map(function(doc) {
return {
updateOne: {
filter: { _id: doc._id },
update: {
$set: {
status: 'processed',
processed_at: new Date()
}
}
}
};
});
var result = collection.bulkWrite(bulkOps);
totalUpdated += result.modifiedCount;
skip += batchSize;
print('已更新: ' + totalUpdated + ' 条文档');
}
print('批量更新完成，总共更新: ' + totalUpdated + ' 条文档');
执行 Shell 脚本
# 执行 MongoDB shell 脚本 mongo batch_update.js
复杂更新逻辑示例
带聚合的批量更新
async function complexBatchUpdate() {
const MongoClient = require('mongodb').MongoClient;
const client = new MongoClient('mongodb://localhost:27017');
await client.connect();
const db = client.db('ecommerce');
const orders = db.collection('orders');
try {
// 准备批量操作
const pipeline = [
{ $match: { status: 'completed', amount: { $gt: 100 } } },
{ $project: {
_id: 1,
discount: { $multiply: ["$amount", 0.1] } // 计算10%折扣
}}
];
const cursor = orders.aggregate(pipeline);
const bulkOps = [];
await cursor.forEach(doc => {
bulkOps.push({
updateOne: {
filter: { _id: doc._id },
update: {
$set: {
discount: doc.discount,
finalAmount: { $subtract: ["$amount", doc.discount] }
}
}
}
});
});
if (bulkOps.length > 0) {
const result = await orders.bulkWrite(bulkOps);
console.log(`成功更新 ${result.modifiedCount} 条订单`);
}
} catch (error) {
console.error('更新失败:', error);
} finally {
await client.close();
}
}
性能优化建议
// 优化批量更新性能
async function optimizedBatchUpdate() {
// 1. 设置合适的批次大小
const BATCH_SIZE = 500; // 不是越大越好，建议500-1000
// 2. 使用 unordered 模式提高性能
const bulkOps = {
ordered: false, // 无序执行，一个失败不影响其他
writeConcern: { w: 1 } // 降低写入确认级别
};
// 3. 索引优化
// 确保更新条件有对应索引
// 4. 使用连接池
const MongoClient = require('mongodb').MongoClient;
const client = new MongoClient('mongodb://localhost:27017', {
poolSize: 10,
useUnifiedTopology: true
});
// 5. 分批提交，避免内存溢出
async function processInBatches(data, batchSize) {
let processed = 0;
while (processed < data.length) {
const batch = data.slice(processed, processed + batchSize);
const bulkOps = batch.map(item => ({
updateOne: {
filter: { _id: item._id },
update: { $set: item.update }
}
}));
await collection.bulkWrite(bulkOps, { ordered: false });
processed += batchSize;
console.log(`进度: ${processed}/${data.length}`);
}
}
}
注意事项
- 备份数据：批量更新前建议备份
- 测试环境：先在测试环境验证
- 监控日志：记录操作日志便于回滚
- 分批处理：避免一次性更新过多文档
- 错误处理：实现重试机制和错误日志
- 性能监控：监控数据库性能指标
选择哪种方法取决于你的具体需求：数据量大小、更新逻辑复杂度、性能要求等。
