26

我使用 mongodb + node.js + mongoose.js ORM 后端。

假设我有一些没有 _id 字段的嵌套对象数组

mongoose.Schema({
  nested: [{
    _id: false, prop: 'string'
  }]
})

然后我想将 _id 字段添加到所有嵌套的对象中，所以猫鼬模式将是

mongoose.Schema({
  nested: [{
    prop: 'string'
  }]
})

然后我应该运行一些脚本来修改生产数据库，对吧？处理这种变化的最佳方法是什么？哪种工具（或方法）最适合用于实施变革？

4

3 回答 3

19

无模式数据库的显着优势之一是您不必使用新的模式布局更新整个数据库。如果数据库中的某些文档没有特定信息，那么您的代码可以改为执行适当的操作，或者选择现在对该记录执行任何操作。

另一种选择是根据需要延迟更新文档 - 仅在再次查看它们时。在这种情况下，您可能会选择使用每个记录/文档的版本标志 - 它最初甚至可能不会出现（因此表示“版本 0”）。即使那是可选的。相反，您的数据库访问代码会查找它需要的数据，如果它不存在，因为它是在代码更新后添加的新信息，那么它将尽其所能填充结果。

对于您的示例，将 an_id:false转换为标准MongoId字段，当代码被读取（或在更新后写回）并且_id:false当前设置时，然后进行更改并仅在绝对需要时才写入。

于 2013-01-21T10:20:13.953 回答
15

您确实必须编写将遍历集合并向每个文档添加新字段的脚本。但是，具体的操作方式取决于数据库的大小和存储系统的性能。在文档中添加一个字段会改变它的大小，因此在大多数情况下会导致重定位。该操作对 IO 有影响，也受其约束。如果您的集合只有几千个文档，可能多达十万个，那么您可以只在一个循环中迭代它，因为整个集合可能适合内存，并且所有 IO 都将在之后发生。但是，如果收集的范围远远超出可用内存，则该方法会更加复杂。我们通常在 MongoDB 的生产使用中遵循以下步骤：

  • 用 timeout=False 打开游标
  • 将一大块文档读入内存
  • 对这些文档运行更新查询
  • 休眠一段时间，避免 IO 子系统过载，影响生产应用
  • 重复直到完成
  • 关闭光标:)

文档块的大小和休眠期必须通过实验确定。通常，您希望在迁移期间避免 mongostats 中的 QR/QW。对于较慢驱动器上的较大集合（如 Amazon 上的 EBS），这种 IO 安全方法可能需要数小时到数天。

于 2013-01-21T09:48:38.653 回答
1

扩展@Michael Korbakov 的答案，我用mongoshell 脚本实现了他的步骤（参见MongoDB 参考手册关于mongoshell 脚本）。

重要提示：如MongoDB 参考手册中所述，在 shell 上运行脚本mongo有助于提高性能，因为它减少了每次批量获取和批量执行的连接延迟。

应该考虑的一个缺点是mongoshell 命令总是同步的，但是批量执行已经为我们处理了并行性（对于每个块），所以我们很适合这个用例。

代码：

// constants
var sourceDbName = 'sourceDb';
var sourceCollectionName = 'sourceColl';
var destDbName = 'destdb';
var destCollectionName = 'destColl';
var bulkWriteChunckSize = 1000;
// for fetching, I figured 1000 for current bulkWrite, and +1000 ready for next bulkWrite
var batchSize = 2000;    
var sourceDb = db.getSiblingDB(sourceDbName);
var destDb = db.getSiblingDB(destDbName);

var start = new Date();

var cursor = sourceDb[sourceCollectionName].find({}).noCursorTimeout().batchSize(batchSize);

var currChunkSize = 0;
var bulk = destDb[destCollectionName].initializeUnorderedBulkOp();
cursor.forEach(function(doc) {
    currChunkSize++;
    bulk.insert({
        ...doc,
        newProperty: 'hello!',
    }); // can be changed for your need, if you want update instead

    if (currChunkSize === bulkWriteChunckSize) {
        bulk.execute();

        // each bulk.execute took for me 130ms, so i figured to wait the same time as well
        sleep(130);

        currChunkSize = 0;
        bulk = destDb[destCollectionName].initializeUnorderedBulkOp();
    }
});

if (currChunkSize > 0) {
    bulk.execute();
    currChunkSize = 0;
}

var end = new Date();
print(end - start);

cursor.close();
于 2020-07-31T17:46:12.230 回答