如何用脚本批量更新MongoDB文档？

本文目录导读：

1. 使用 bulkWrite() 方法（推荐）
2. 基于条件的批量更新
3. 逐条处理的大批量更新
4. Shell 脚本批量更新
5. 复杂更新逻辑示例
6. 性能优化建议
7. 注意事项

我来详细介绍几种批量更新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}`);
    }
  }
}

注意事项

1. 备份数据：批量更新前建议备份
2. 测试环境：先在测试环境验证
3. 监控日志：记录操作日志便于回滚
4. 分批处理：避免一次性更新过多文档
5. 错误处理：实现重试机制和错误日志
6. 性能监控：监控数据库性能指标

选择哪种方法取决于你的具体需求：数据量大小、更新逻辑复杂度、性能要求等。