Inserimento bulk documentdb di Azure mediante stored procedure

Question

Ciao Sto usando 16 raccolte per inserire circa 3-4 milioni di oggetti JSON che vanno da 5-10k per oggetto. Sto usando la stored procedure per inserire questi documenti. Ho 22 unità di capacità.

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); 
     } 
    } 

miei C# codici appare come questa

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; 
      } 

Il problema che sto affrontando è fuori 400k documenti che cerco di inserire a volte documenti inevasa con fuori dando alcun errore.

L'applicazione è un ruolo di lavoro distribuito su cloud. Se aumento il numero di thread o istanze che si inseriscono in documentDB, il numero di documenti persi è molto più alto.

come capire qual è il problema.Grazie in anticipo.

Answer 1

È importante notare che le stored procedure hanno un'esecuzione limitata, in cui tutte le operazioni devono essere completate entro la durata del timeout della richiesta specificata dal server. Se un'operazione non viene completata con quel limite di tempo, la transazione viene automaticamente ripristinata. Per semplificare lo sviluppo per gestire i limiti di tempo, tutte le operazioni CRUD (Crea, Leggi, Aggiorna ed Elimina) restituiscono un valore booleano che indica se l'operazione verrà completata. Questo valore booleano può essere utilizzato come segnale per concludere l'esecuzione e per implementare un modello basato sulla continuazione per riprendere l'esecuzione (questo è illustrato nei nostri esempi di codice di seguito).

La stored procedure bulk-insert sopra riportata implementa il modello di continuazione restituendo il numero di documenti creati correttamente. Questo si nota nei commenti della stored procedure:

// 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); 

Se il conteggio documento di output è inferiore al numero di documento di input, sarà necessario eseguire nuovamente la stored procedure con il set restante di documenti.

Answer 2

Ho scoperto che durante il tentativo di questo codice avrei ricevuto un errore su docs.length che indicava che la lunghezza non era definita.

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 
} 

Dopo molte prove (non riusciva a trovare nulla nella documentazione Azure) mi sono reso conto che non potevo passare un array come è stato suggerito. Il parametro doveva essere un oggetto. Ho dovuto modificare il codice batch in questo modo per poterlo eseguire.

Ho anche scoperto che non potevo semplicemente provare a passare una serie di documenti nello script explorer di DocumentDB (casella di input). Anche se il testo di aiuto segnaposto dice che puoi.

Questo codice ha funzionato per me:

// 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. 
} 

documentazione Speriamo Azure di recuperare il ritardo o diventare più facile da trovare se ho perso.

Inserirò anche una segnalazione di bug per Script Explorer nella speranza che gli Azuriti si aggiornino.