Jaký je správný přístup k aktualizaci mnoha záznamů v MongoDB pomocí Mongoose

Přístup spočívající v sestavení kritéria sestávajícího ze všech ID dokumentů a následném provedení aktualizace nutně způsobí potenciální problémy. Když iterujete seznam dokumentů a odesíláte aktualizační operaci s každým dokumentem, v Mongoose riskujete, že vyhodíte váš server do povětří, zejména při práci s velkou datovou sadou, protože nečekáte na dokončení asynchronního volání, než přejdete k dalšímu. opakování. V podstatě budete vytvářet "hromadu" nevyřešených operací, dokud to nezpůsobí problém - Stackoverflow.

Vezměme si například, že máte pole ID dokumentů, které chcete aktualizovat odpovídající dokument ve stavovém poli:

const processedIds = [
  "57a0a96bd1c6ef24376477cd",
  "57a052242acf5a06d4996537",
  "57a052242acf5a06d4996538"
];

kde můžete použít updateMany() metoda

Model.updateMany(
  { _id: { $in: processedIds } }, 
  { $set: { status: "processed" } }, 
  callback
);

nebo alternativně pro opravdu malé datové sady můžete použít forEach() metodu na poli pro iteraci a aktualizaci vaší kolekce:

processedIds.forEach(function(id)){
  Model.update({ _id: id}, { $set: { status: "processed" } }, callback);
});

Výše uvedené je v pořádku pro malé datové sady. To se však stává problémem, když čelíte tisícům nebo milionům dokumentů, které je třeba aktualizovat, protože v rámci smyčky budete opakovaně volat server asynchronního kódu.

Chcete-li to překonat, použijte něco jako asynchronní eachLimit a iterujte pole a provádějte operaci aktualizace MongoDB pro každou položku, přičemž nikdy neprovádějte více než x paralelních aktualizací ve stejnou dobu.

Nejlepším přístupem by bylo použít k tomu hromadné API, které je extrémně efektivní při hromadném zpracování aktualizací. Rozdíl ve výkonu oproti volání operace aktualizace na každém z mnoha dokumentů spočívá v tom, že namísto odesílání požadavků na aktualizaci na server s každou iterací hromadné API odesílá požadavky jednou za každých 1000 požadavků (dávkově).

Pro verze Mongoose >=4.3.0 které podporují MongoDB Server 3.2.x , můžete použít bulkWrite() pro aktualizace. Následující příklad ukazuje, jak toho můžete dosáhnout:

const bulkUpdateCallback = function(err, r){
  console.log(r.matchedCount);
  console.log(r.modifiedCount);
}

// Initialize the bulk operations array
const bulkUpdateOps = [], counter = 0;

processedIds.forEach(function (id) {
  bulkUpdateOps.push({
    updateOne: {
      filter: { _id: id },
      update: { $set: { status: "processed" } }
    }
  });
  counter++;

  if (counter % 500 == 0) {
    // Get the underlying collection via the Node.js driver collection object
    Model.collection.bulkWrite(bulkUpdateOps, { ordered: true, w: 1 }, bulkUpdateCallback);
    bulkUpdateOps = []; // re-initialize
  }
})

// Flush any remaining bulk ops
if (counter % 500 != 0) {
  Model.collection.bulkWrite(bulkOps, { ordered: true, w: 1 }, bulkUpdateCallback);
}

Pro verze Mongoose ~3.8.8 , ~3.8.22 , 4.x které podporují MongoDB Server >=2.6.x , můžete použít Bulk API následovně

var bulk = Model.collection.initializeOrderedBulkOp(),
    counter = 0;

processedIds.forEach(function(id) {
    bulk.find({ "_id": id }).updateOne({ 
        "$set": { "status": "processed" }
    });

    counter++;
    if (counter % 500 == 0) {
        bulk.execute(function(err, r) {
           // do something with the result
           bulk = Model.collection.initializeOrderedBulkOp();
           counter = 0;
        });
    }
});

// Catch any docs in the queue under or over the 500's
if (counter > 0) {
    bulk.execute(function(err,result) {
       // do something with the result here
    });
}