Introduction
Using the .NET SDK, it’s fast and easy to store individual documents in an Azure Cosmos DB container, where typically, each write operation completes in under 10 milliseconds.
However, the .NET SDK can also support bulk operations, for scenarios where you need to load large volumes of data, with as much throughput as possible. Like, imagine you need to dump two million documents into a container – that’s bulk. And while bulk inserts are most common, bulk updates and deletes are also supported. On the other hand, if you’re running individual point operations that need to complete as quickly as possible, then it would not be appropriate to issue those using bulk execution.
It’s very easy to enable bulk execution in your applications. First, set AllowBulkExecution to true in your Cosmos client constructor. Then, populate a list of tasks, one for each operation. Finally, you just run Task.WhenAll on the list, and the rest just happens like magic.
For example, you create a single document by calling await CreateItemAsync. Well, if you’ve got a thousand documents to create, you just create a list of a thousand tasks. Then you call the same CreateItemAsync method on each, only without the await keyword – and that returns a task for the operation, without running it. Finally, calling await Task.WhenAll on the entire list executes the bulk operation much more efficiently than doing it one at a time, and as you’ll see in this post, yields dramatic performance gains.
You’ll also need to handle exceptions of course, since each task is still an individual operation that can succeed or fail on its own. For a single insert, we’d use a typical try/catch block, but there’s a different pattern for bulk execution. When you add each task to the list, task on a call to ContinueWith, and that lets you run any code you want after each task completes. Instead of a try/catch, you’ll get handed back the same task object that just ran. And that object has a Status property, which you can test for Faulted, meaning that an unhandled exception occurred causing the task to fail. In that case, you can get the exception from the task’s Exception property, and handle it any way that you need to.
Again, it all just happens like magic, thanks to the .NET SDK, which – internally – groups concurrent operations on the client, and then dispatches a single request for each group. As a result, the client issues far fewer requests then actual documents – which yields far greater performance for bulk operations. Let’s see bulk execution in action.
Generating Documents
To demonstrate, we’ll use a test container that has throughput provisioned at 10,000 RUs (request units) a second, which is fairly high. Issuing individual write operations one at a time on a single thread, we’d never come anywhere near utilizing all that throughput. But with bulk execution, the .NET SDK will attempt to saturate all that available throughput.
First, let’s write a GenerateItems method to create an array of documents in memory for writing to the container:
private static Item[] GenerateItems(int count)
{
var items = new Item[count];
for (var i = 0; i < count; i++)
{
var id = Guid.NewGuid().ToString();
items[i] = new Item
{
id = id,
pk = id,
username = $"user{i}"
};
}
// Simulate a duplicate to cause an error inserting a document
items[1].id = items[0].id;
items[1].pk = items[0].pk;
return items;
}
All we’re doing here is creating an array of Item objects, which have ID, PK (partition key), and Username properties. And to help test our exception handling, we’ll simulate a duplicate by copying the ID and PK values from the first element into the second element. This means that second document will fail, because you can’t have two documents with the same ID and partition key in one container. So if we generate 100 documents for example, we should really expect only 99 documents the get created successfully.
Without Bulk Execution
We’ll load documents into the container twice – first, one at a time, and then again using bulk execution. Here’s the code to perform the writes one at a time (not using bulk execution):
var items = GenerateItems(count);
var cost = 0D;
var errors = 0;
var started = DateTime.Now;
var container = Shared.Client.GetContainer("adventure-works", "bulkdemo");
foreach (var item in items)
{
try
{
var result = await container.CreateItemAsync(item, new PartitionKey(item.pk));
cost += result.RequestCharge;
}
catch (Exception ex)
{
Console.WriteLine($"Error creating document: {ex.Message}");
errors++;
}
}
Console.WriteLine($"Created {count - errors} documents (non-bulk): {cost:0.##} RUs in {DateTime.Now.Subtract(started)}");
This code loops through the items array, and creates a document in Cosmos DB for each item. For each item, we simply await on a call to CreateItemAsync for each document, and check for errors using a try/catch. So let’s run that for 100 documents.
We see the Conflict (409) error caused by the duplicate and displayed by our catch block. And when it’s done, we can see that we successfully inserted only 99 documents, in about 10 seconds, at a cost of 622 RUs.
With Bulk Execution
OK, now let’s do the same thing using bulk execution. First, remember we need to enable bulk execution in our Cosmos client, which is defined by a shared Client property:
public static class Shared
{
public static CosmosClient Client { get; private set; }
static Shared()
{
var config = new ConfigurationBuilder().AddJsonFile("appsettings.json").Build();
var endpoint = config["CosmosEndpoint"];
var masterKey = config["CosmosMasterKey"];
Client = new CosmosClient(endpoint, masterKey,
new CosmosClientOptions { AllowBulkExecution = true });
}
}
Here you can see that, inside the Cosmos client constructor, we’re setting AllowBulkExecution to true.
With bulk execution enabled, we now instantiate and populate a list of tasks:
var items = GenerateItems(count);
var cost = 0D;
var errors = 0;
var started = DateTime.Now;
var container = Shared.Client.GetContainer("adventure-works", "bulkdemo");
var tasks = new List<Task>(count);
foreach (var item in items)
{
var task = container.CreateItemAsync(item, new PartitionKey(item.pk));
tasks.Add(task
.ContinueWith(t =>
{
if (t.Status == TaskStatus.RanToCompletion)
{
cost += t.Result.RequestCharge;
}
else
{
Console.WriteLine($"Error creating document: {t.Exception.Message}");
errors++;
}
}));
}
await Task.WhenAll(tasks);
Console.WriteLine($"Created {count - errors} documents (bulk): {cost:0.##} RUs in {DateTime.Now.Subtract(started)}");
For each document, we call CreateItemAsync as before, only without the await. This gives us the task for the document, which we add to the list, and call ContinueWith so we can check on the success of each individual operation. If the task isn’t faulted, we track the RU charge, otherwise and exception occurred. So the else block inside ContinueWith is kind of your catch block when you’re doing bulk execution. The faulted task gives us the actual exception that occurred in its Exception property, which we just write out to the console.
Running this code, you can see how the bulk execution dramatically boosts performance:
We see the same single failure for the duplicate document, with 99 documents being created in just .14 seconds. Also notice that the bulk operations also lowered the throughput cost slightly, down from 622 RUs to 591.
That’s .14 seconds using bulk execution, compared to 10 seconds without.
Now let’s run the demo again, this time with 1,000 documents:
Well, there’s the same error for that duplicate, as the documents get created one at a time, until finally, we see that 999 documents got created in a minute and 40 seconds, at a cost of 6,283 RUs.
Finally, let’s compare that to bulk execution for another 1,000 documents:
Hard to believe, but we just did the same thing in about half a second, again at a slightly lower cost of about 6,088 RUs.
As you can see, the performance gains are dramatic, so make sure to leverage bulk execution when you need to load large amounts of data into Cosmos DB from your .NET applications.
Happy coding!
Lenni's Technology Blog 