Azure Documentdb Masseneinfügung mit gespeicherten Prozedur

Question

Hallo Ich verwende 16 Sammlungen, um etwa 3-4 Millionen JSON-Objekte im Bereich von 5-10k pro Objekt einzufügen.Ich verwende gespeicherte Prozedur, um diese Dokumente einzufügen.Ich habe 22 Capacity Unit.

function bulkImport(docs) { 
    var collection = getContext().getCollection(); 
    var collectionLink = collection.getSelfLink(); 

    // The count of imported docs, also used as current doc index. 
    var count = 0; 

    // Validate input. 
    if (!docs) throw new Error("The array is undefined or null."); 

    var docsLength = docs.length; 
    if (docsLength == 0) { 
     getContext().getResponse().setBody(0); 
    } 

    // Call the CRUD API to create a document. 
    tryCreateOrUpdate(docs[count], callback); 

    // Note that there are 2 exit conditions: 
    // 1) The createDocument request was not accepted. 
    // In this case the callback will not be called, we just call setBody and we are done. 
    // 2) The callback was called docs.length times. 
    // In this case all documents were created and we don't need to call tryCreate anymore. Just call setBody and we are done. 
    function tryCreateOrUpdate(doc, callback) { 
     var isAccepted = true; 
     var isFound = collection.queryDocuments(collectionLink, 'SELECT * FROM root r WHERE r.id = "' + doc.id + '"', function (err, feed, options) { 
      if (err) throw err; 
      if (!feed || !feed.length) { 
       isAccepted = collection.createDocument(collectionLink, doc, callback); 
      } 
      else { 
       // The metadata document. 
       var existingDoc = feed[0]; 
       isAccepted = collection.replaceDocument(existingDoc._self, doc, callback); 
      } 
     }); 

     // If the request was accepted, callback will be called. 
     // Otherwise report current count back to the client, 
     // which will call the script again with remaining set of docs. 
     // This condition will happen when this stored procedure has been running too long 
     // and is about to get cancelled by the server. This will allow the calling client 
     // to resume this batch from the point we got to before isAccepted was set to false 
     if (!isFound && !isAccepted) getContext().getResponse().setBody(count); 
    } 

    // This is called when collection.createDocument is done and the document has been persisted. 
    function callback(err, doc, options) { 
     if (err) throw err; 

     // One more document has been inserted, increment the count. 
     count++; 

     if (count >= docsLength) { 
      // If we have created all documents, we are done. Just set the response. 
      getContext().getResponse().setBody(count); 
     } else { 
      // Create next document. 
      tryCreateOrUpdate(docs[count], callback); 
     } 
    } 

meine C# Codes sieht wie folgt aus

public async Task<int> Add(List<JobDTO> entities) 
      { 

        int currentCount = 0; 
        int documentCount = entities.Count; 

        while(currentCount < documentCount) 
        { 
         string argsJson = JsonConvert.SerializeObject(entities.Skip(currentCount).ToArray()); 
         var args = new dynamic[] { JsonConvert.DeserializeObject<dynamic[]>(argsJson) }; 

         // 6. execute the batch. 
         StoredProcedureResponse<int> scriptResult = await DocumentDBRepository.Client.ExecuteStoredProcedureAsync<int>(sproc.SelfLink, args); 

         // 7. Prepare for next batch. 
         int currentlyInserted = scriptResult.Response; 

         currentCount += currentlyInserted; 

        } 

        return currentCount; 
      } 

Das Problem, das ich bin vor aus 400k Dokumenten ist, dass ich versuche manchmal Dokumente einfügen lassen verpassten ohne jede Störung zu geben.

Die Anwendung ist eine in der Cloud bereitgestellte Worker-Rolle. Wenn ich die Anzahl der Threads oder Instanzen in DocumentDB erhöhen, ist die Anzahl der verpassten Dokumente viel höher.

wie herauszufinden, was das Problem ist. Vielen Dank im Voraus.

Answer 1

Es ist wichtig zu beachten, dass gespeicherte Prozeduren eine eingeschränkte Ausführung haben, bei der alle Operationen innerhalb der vom Server angegebenen Anforderungs-Timeoutdauer abgeschlossen sein müssen. Wenn eine Operation mit diesem Zeitlimit nicht abgeschlossen wird, wird die Transaktion automatisch zurückgesetzt. Um die Entwicklung zu vereinfachen, um Zeitlimits zu verarbeiten, geben alle CRUD-Operationen (Create, Read, Update und Delete) einen booleschen Wert zurück, der angibt, ob diese Operation abgeschlossen wird. Dieser boolesche Wert kann als Signal verwendet werden, um die Ausführung abzuschließen und ein fortsetzungsbasiertes Modell zu implementieren, um die Ausführung fortzusetzen (dies wird in unseren Codebeispielen unten veranschaulicht).

Die oben zur Verfügung gestellte gespeicherte Prozedur für die Masseneinfügung implementiert das Fortsetzungsmodell, indem die Anzahl der erfolgreich erstellten Dokumente zurückgegeben wird. Dies wird in der gespeicherten Prozedur Anmerkungen notiert:

// If the request was accepted, callback will be called. 
    // Otherwise report current count back to the client, 
    // which will call the script again with remaining set of docs. 
    // This condition will happen when this stored procedure has been running too long 
    // and is about to get cancelled by the server. This will allow the calling client 
    // to resume this batch from the point we got to before isAccepted was set to false 
    if (!isFound && !isAccepted) getContext().getResponse().setBody(count); 

Wenn das Ausgabedokument Zählung kleiner als die Eingangsdokumentenanzahl, müssen Sie die gespeicherte Prozedur mit dem verbleibenden Satz von Dokumenten erneut ausführen.

Answer 2

Ich fand, dass, wenn ich diesen Code versuchte, ich einen Fehler bei docs.length erhalten würde, der besagte, dass Länge undefined war.

function bulkImport(docs) { 
    var collection = getContext().getCollection(); 
    var collectionLink = collection.getSelfLink(); 

    // The count of imported docs, also used as current doc index. 
    var count = 0; 

    // Validate input. 
    if (!docs) throw new Error("The array is undefined or null."); 

    var docsLength = docs.length; // length is undefined 
} 

Nach vielen Tests (irgendetwas nicht in Azure Dokumentation finden konnte) wurde mir klar, dass ich nicht ein Array übergeben konnte, wie vorgeschlagen wurde. Der Parameter muss ein Objekt sein. Ich musste den Stapelcode so ändern, dass er ausgeführt wird.

Ich fand auch, dass ich nicht einfach versuchen konnte, ein Array von Dokumenten im DocumentDB-Skript-Explorer (Eingabefeld) zu übergeben. Auch wenn der Platzhalter Hilfetext sagt, können Sie.

Dieser Code funktioniert für mich:

// psuedo object for reference only 
docObject = { 
    "items": [{doc}, {doc}, {doc}] 
} 

function bulkImport(docObject) { 
    var context = getContext(); 
    var collection = context.getCollection(); 
    var collectionLink = collection.getSelfLink(); 
    var count = 0; 

    // Check input 
    if (!docObject.items || !docObject.items.length) throw new Error("invalid document input parameter or undefined."); 
    var docs = docObject.items; 
    var docsLength = docs.length; 
    if (docsLength == 0) { 
     context.getResponse().setBody(0); 
    } 

    // Call the funct to create a document. 
    tryCreateOrUpdate(docs[count], callback); 

    // Obviously I have truncated this function. The above code should help you understand what has to change. 
} 

Hoffentlich Azure Dokumentation aufholen oder einfacher werden, wenn ich es verpasst zu finden.

Ich werde auch einen Fehlerbericht für den Skript Explorer in der Hoffnung, dass die Azurites aktualisieren werden.