Dela via

Migrera från massexekutorbiblioteket till masssupporten i Azure Cosmos DB .NET V3 SDK

GÄLLER FÖR: NoSQL

Den här artikeln beskriver de steg som krävs för att migrera ett befintligt programs kod som använder .NET-massexekutorbiblioteket till masssupportfunktionen i den senaste versionen av .NET SDK.

Aktivera massstöd

Aktivera massstöd för instansen CosmosClientvia AllowBulkExecution-konfigurationen :

new CosmosClient(endpoint, authKey, new CosmosClientOptions() { AllowBulkExecution = true });

Skapa uppgifter för varje åtgärd

Massstöd i .NET SDK fungerar genom att använda det parallella aktivitetsbiblioteket och grupperingsåtgärder som utförs samtidigt.

Det finns ingen enskild metod i SDK som tar din lista över dokument eller åtgärder som en indataparameter, utan du behöver skapa en aktivitet för varje åtgärd som du vill köra i grupp och sedan bara vänta tills de har slutförts.

Om dina första indata till exempel är en lista över objekt där varje objekt har följande schema:

public class MyItem
{
    public string id { get; set; }

    public string pk { get; set; }

    public int operationCounter { get; set; } = 0;
}

Om du vill massimportera (liknar att använda BulkExecutor.BulkImportAsync) måste du ha samtidiga anrop till CreateItemAsync. Till exempel:

BulkOperations<MyItem> bulkOperations = new BulkOperations<MyItem>(documentsToWorkWith.Count);
foreach (MyItem document in documentsToWorkWith)
{
    bulkOperations.Tasks.Add(CaptureOperationResponse(container.CreateItemAsync(document, new PartitionKey(document.pk)), document));
}

Om du vill göra massuppdateringar (ungefär som med BulkExecutor.BulkUpdateAsync) måste du ha samtidiga anrop till ReplaceItemAsync metoden när du har uppdaterat objektvärdet. Till exempel:

BulkOperations<MyItem> bulkOperations = new BulkOperations<MyItem>(documentsToWorkWith.Count);
foreach (MyItem document in documentsToWorkWith)
{
    document.operationCounter++;
    bulkOperations.Tasks.Add(CaptureOperationResponse(container.ReplaceItemAsync(document, document.id, new PartitionKey(document.pk)), document));
}

Och om du vill göra massradering (liknar att använda BulkExecutor.BulkDeleteAsync) behöver du ha samtidiga anrop till DeleteItemAsync, med id och partitionsnyckeln för varje objekt. Till exempel:

BulkOperations<MyItem> bulkOperations = new BulkOperations<MyItem>(documentsToWorkWith.Count);
foreach (MyItem document in documentsToWorkWith)
{
    document.operationCounter++;
    bulkOperations.Tasks.Add(CaptureOperationResponse(container.DeleteItemAsync<MyItem>(document.id, new PartitionKey(document.pk)), document));
}

Resultattillstånd för avbildningsaktivitet

I föregående kodexempel har vi skapat en samtidig lista över uppgifter och anropat CaptureOperationResponse metoden för var och en av dessa uppgifter. Den här metoden är ett tillägg som gör att vi kan upprätthålla ett liknande svarsschema som BulkExecutor genom att samla in eventuella fel och spåra användningen av enheter för begäranden.

private static async Task<OperationResponse<T>> CaptureOperationResponse<T>(Task<ItemResponse<T>> task, T item)
{
    try
    {
        ItemResponse<T> response = await task;
        return new OperationResponse<T>()
        {
            Item = item,
            IsSuccessful = true,
            RequestUnitsConsumed = task.Result.RequestCharge
        };
    }
    catch (Exception ex)
    {
        if (ex is CosmosException cosmosException)
        {
            return new OperationResponse<T>()
            {
                Item = item,
                RequestUnitsConsumed = cosmosException.RequestCharge,
                IsSuccessful = false,
                CosmosException = cosmosException
            };
        }

        return new OperationResponse<T>()
        {
            Item = item,
            IsSuccessful = false,
            CosmosException = ex
        };
    }
}

Där OperationResponse deklareras som:

public class OperationResponse<T>
{
    public T Item { get; set; }
    public double RequestUnitsConsumed { get; set; } = 0;
    public bool IsSuccessful { get; set; }
    public Exception CosmosException { get; set; }
}

Utföra operationer samtidigt

För att spåra omfånget för hela listan med uppgifter använder vi den här hjälpklassen:

public class BulkOperations<T>
{
    public readonly List<Task<OperationResponse<T>>> Tasks;

    private readonly Stopwatch stopwatch = Stopwatch.StartNew();

    public BulkOperations(int operationCount)
    {
        this.Tasks = new List<Task<OperationResponse<T>>>(operationCount);
    }

    public async Task<BulkOperationResponse<T>> ExecuteAsync()
    {
        await Task.WhenAll(this.Tasks);
        this.stopwatch.Stop();
        return new BulkOperationResponse<T>()
        {
            TotalTimeTaken = this.stopwatch.Elapsed,
            TotalRequestUnitsConsumed = this.Tasks.Sum(task => task.Result.RequestUnitsConsumed),
            SuccessfulDocuments = this.Tasks.Count(task => task.Result.IsSuccessful),
            Failures = this.Tasks.Where(task => !task.Result.IsSuccessful).Select(task => (task.Result.Item, task.Result.CosmosException)).ToList()
        };
    }
}

Metoden ExecuteAsync väntar tills alla åtgärder har slutförts och du kan använda den så här:

BulkOperationResponse<MyItem> bulkOperationResponse = await bulkOperations.ExecuteAsync();

Fånga statistik

Den tidigare koden väntar tills alla åtgärder har slutförts och beräknar den statistik som krävs. Den här statistiken liknar den för massexekutorbibliotekets BulkImportResponse.

public class BulkOperationResponse<T>
{
    public TimeSpan TotalTimeTaken { get; set; }
    public int SuccessfulDocuments { get; set; } = 0;
    public double TotalRequestUnitsConsumed { get; set; } = 0;

    public IReadOnlyList<(T, Exception)> Failures { get; set; }
}

Det BulkOperationResponse innehåller:

  1. Den totala tid det tar att bearbeta listan över åtgärder via massstöd.
  2. Antalet lyckade åtgärder.
  3. Det totala antalet förbrukade begärningsenheter.
  4. Om det uppstår fel visas en lista över tupplar som innehåller undantaget och det associerade objektet för loggning och identifiering.

Försök konfigurera igen

Massexekutorbiblioteket hade vägledning som nämnde att ange MaxRetryWaitTimeInSeconds och MaxRetryAttemptsOnThrottledRequests för RetryOptions till 0 för att delegera kontroll till biblioteket.

För massstöd i .NET SDK finns det inget dolt beteende. Du kan konfigurera alternativen för återförsök direkt genom CosmosClientOptions.MaxRetryAttemptsOnRateLimitedRequests och CosmosClientOptions.MaxRetryWaitTimeOnRateLimitedRequests.

Kommentar

Om enheterna för etablerade begäranden är mycket lägre än förväntat baserat på mängden data kan du överväga att ställa in dessa till höga värden. Massåtgärden tar längre tid, men den har större chans att lyckas helt tack vare fler återförsök.

Prestandaförbättringar

Precis som med andra åtgärder med .NET SDK ger användning av stream-API:er bättre prestanda och undviker onödig serialisering.

Det går bara att använda stream-API:er om den typ av data du använder matchar en ström av bytes (till exempel filströmmar). I sådana fall ökar användningen av metoderna CreateItemStreamAsync, ReplaceItemStreamAsync eller DeleteItemStreamAsync och att arbeta med ResponseMessage (i stället för ItemResponse) det genomflöde som kan uppnås.

Nästa steg