Handling Error 429 in DocumentDB

I am working on one of these IoT projects at the moment and one of my requirements is to stream events from devices into DocumentDB.  My simplified architecture looks like this. 

 

Event Hub to Worker Role to DocumentDB

 

 

Each collection in DocumentDB is provisioned with a certain amount of throughput (RU == Request Units).  If you exceed that amount of requests you will receive back a 429 error “Too Many Request“. To get around this you can

 

1. Move your collection to a more performant level (S1, S2, S3)
2. Implement throttling retries
3. Tune the requests to need less RUs
4. Implement Partitions

#1 makes sense because if you are hitting DocumentDB that hard then you need to get the right “SKU” to tackle the job (but what if S3 is not enough)
#2 means capture the fact you hit this error and retry (I’ll show an example)
#3 in an IoT scenario could be taken to mean “Slow down the sensors”, but also includes indexing strategy (change to lazy indexing and/or excluding paths)
#4 IMHO the best and most logical option. (allows you to scale out.  Brilliant)

 

Scenario

My scenario is that I want to take the cheapest option so even though I think #4 is the right option, it will cost me money for each collection.  I want to take a look at #2.  I want to show you what I think is the long way around and then show you the “Auto retry” option.

 

The Verbose Route

 

        public async Task SaveToDocDb(string uri, string key, dynamic jsonDocToSave)
        {
 
            using (var client = new DocumentClient(new Uri(uri), key))
            {
                var queryDone = false;
                while (!queryDone)
                {
                    try
                    {
                        await client.CreateDocumentAsync(coll.SelfLink, jsonDocToSave);
                        queryDone = true;
                    }
                    catch (DocumentClientException documentClientException)
                    {
                        var statusCode = (int)documentClientException.StatusCode;
                        if (statusCode == 429) 
                            Thread.Sleep(documentClientException.RetryAfter);
                        //add other error codes to trap here e.g. 503 - Service Unavailable
						else
                            throw;
                    }
                    catch (AggregateException aggregateException) when (aggregateException is DocumentClientException )
                    {
                            var statusCode = (int)aggregateException.StatusCode;
                            if (statusCode == 429)
                                Thread.Sleep(aggregateException.RetryAfter);
							//add other error codes to trap here e.g. 503 - Service Unavailable
                            else
                                throw;
                        }
                    }
                }
            }
        }

 

 

The above is a very common pattern TRY…CATCH, get the error number and take action.  I wanted something less verbose and would have automatic retry logic that made sense. 

 

The Less Verbose Route

 

To do it I had to add another package to my project.  In the Nuget Package Manager Console

Install-Package Microsoft.Azure.DocumentDB.TransientFaultHandling

 

**
This is not a DocumentDB team library and it is unclear as to whether it is still being maintained.  What is clear is that the DocumentDB team will be bringing this retry logic into the SDK natively.  This will mean a lighter more consistent experience and no reliance on an external library.
**

Now add some using statements to the project

using Microsoft.Azure.Documents.Client.TransientFaultHandling;
using Microsoft.Azure.Documents.Client.TransientFaultHandling.Strategies;
using Microsoft.Practices.EnterpriseLibrary.TransientFaultHandling;

And now instead of instantiating a regular DocumentClient in your application you can do this.

 

private IReliableReadWriteDocumentClient CreateClient(string uri, string key)
{
ConnectionPolicy policy = new ConnectionPolicy()
{
ConnectionMode = ConnectionMode.Direct,
ConnectionProtocol = Protocol.Tcp
};
var documentClient = new DocumentClient(new Uri(uri), key,policy);
var documentRetryStrategy = new DocumentDbRetryStrategy(RetryStrategy.DefaultExponential) { FastFirstRetry = true };
return documentClient.AsReliable(documentRetryStrategy);
} 

 

Summary

You may never hit the upper ends of the Request Units in your collection but in an IoT scenario like this or doing large scans over a huge collection of documents  you may hit this error and you need to know how to deal with it.  This article has provided you with two ways to handle this need.  Enjoy

 

My thanks go to Ryan Crawcour of the DocumentDB team for proof reading and sanity checking

Deleting Multiple Documents from Azure DocumentDB

I have been using DocumentDB a lot recently and wanted to share with you something that is harder to do than it should be.  When I say harder I mean you have to type more to get what seems a really easy thing to do.  I will also tell you how to do this more efficiently but because of cost sensitivity I couldn’t do it that way.

Scenario

I have a collection of documents.  These documents are being streamed into DocumentDB from a worker role which is reading from Azure Event Hubs.  As you can imagine I get a lot of documents in a relatively short space of time.  The size of a DocumentDB collection is 10GB.  I only want to use one collection.  My situation is that really I only need to keep two days worth of data in the collection at any one time.  My requirements therefore are

• Maintain only one collection
• Retain only two days worth of data
• Remove documents on a schedule.

For point #3 I am using a .net StopWatch object and that is really simple.  Having only one collection is also very simple so really it comes down to

How do I delete a bunch of documents from DocumentDB?

First Attempt

//setup
string databaseId = ConfigurationManager.AppSettings["DatabaseId"];
string collectionId = ConfigurationManager.AppSettings["CollectionId"];
string endpointUrl = ConfigurationManager.AppSettings["EndPointUrl"];
string authorizationKey = ConfigurationManager.AppSettings["AuthorizationKey"];
//connection policy
ConnectionPolicy policy = new ConnectionPolicy()
{
ConnectionMode = ConnectionMode.Direct,
ConnectionProtocol = Protocol.Tcp
};
//Build our selection criteria
var sqlquery = "SELECT * FROM c WHERE " + ToUnixTime(DateTime.Now).ToString() + " - c.time > 172800";
//Get our client
DocumentClient client = new DocumentClient(new Uri(endpointUrl), authorizationKey, policy);
//Database
Database database = client.CreateDatabaseQuery().Where(db => db.Id == databaseId).ToArray().FirstOrDefault();

//Get a reference to the collection
DocumentCollection coll = client.CreateDocumentCollectionQuery(database.SelfLink).Where(c => c.Id == collectionId).ToArray().FirstOrDefault();
//Issue our query against the collection
var results = client.CreateDocumentQuery<Document>(coll.DocumentsLink, sqlquery).AsEnumerable();
Console.WriteLine("Deleting Documents");
//How many documents do we have to delete
Console.WriteLine("Count of docs to delete = {0}", results.Count().ToString());
//Enumerate the collection
foreach (var item in results)
{
// Console.WriteLine("Deleting");
client.DeleteDocumentAsync(item.SelfLink);
}
//How many documents are still left
var postquery = client.CreateDocumentQuery<Document>(coll.DocumentsLink, sqlquery).AsEnumerable();
Console.WriteLine("Count of docs remaining = {0}", postquery.Count().ToString());
Console.ReadLine();

You may be expecting the result of the second count to be 0.  Unless you have 100 documents or less as the result of the first query then you are going to be disappointed.  We enumerate through the result of our first query getting a reference to each document and deleting it.  Seem fine?  The problem is that DocumentDB only returns 100 documents to us at a time and we didn’t go back and ask for more.  The solution is to execute our query and tell DocumentDB to re-execute the query if it has more results.  You can see a visual example of this when you use Query Explorer in the portal.  Down the bottom, under your query after execution you will find something like this.

The solution.

Here is the code that asks if there are more results to be had and if there are then can we go get the next batch

//setup
string databaseId = ConfigurationManager.AppSettings["DatabaseId"];
string collectionId = ConfigurationManager.AppSettings["CollectionId"];
string endpointUrl = ConfigurationManager.AppSettings["EndPointUrl"];
string authorizationKey = ConfigurationManager.AppSettings["AuthorizationKey"];
//Connection Policy
ConnectionPolicy policy = new ConnectionPolicy()
{
ConnectionMode = ConnectionMode.Direct,
ConnectionProtocol = Protocol.Tcp
};
//build our selection criteria
var sqlquery = "SELECT * FROM c WHERE " + ToUnixTime(DateTime.Now).ToString() + " - c.time > 172800";
//client
DocumentClient client = new DocumentClient(new Uri(endpointUrl), authorizationKey, policy);
//database
Database database = client.CreateDatabaseQuery().Where(db => db.Id == databaseId).ToArray().FirstOrDefault();

//Get a reference to the collection
DocumentCollection coll =
client.CreateDocumentCollectionQuery(database.SelfLink)
.Where(c => c.Id == collectionId)
.ToArray()
.FirstOrDefault();
//First execution of the query
var results = client.CreateDocumentQuery<Document>(coll.DocumentsLink, sqlquery).AsDocumentQuery();

Console.WriteLine("Deleting Documents");
//While there are more results
while (results.HasMoreResults)
{
Console.WriteLine("Has more...");
//enumerate and delete the documents in this batch
foreach (Document doc in await results.ExecuteNextAsync())
{
await client.DeleteDocumentAsync(doc.SelfLink);
}
}
//second count should now be 0
var postquery = client.CreateDocumentQuery<Document>(coll.DocumentsLink, sqlquery).AsEnumerable();
Console.WriteLine("Count of docs remaining = {0}", postquery.Count().ToString());

The key is this statement

var results = client.CreateDocumentQuery<Document>(coll.DocumentsLink, sqlquery).AsDocumentQuery();

using AsDocumentQuery allows us to know if we have more results.

The Easy Way

The easy way and most definitely the proper way to do this is to use Partitions in DocumentDB.  A Partition is essentially a collection.  There are different types of partition but for this example I would have used a range partition over time.  When I wanted to delete documents I would have just simply dropped a partition.  My partitions would have been based on dates.  I would always have had 2 full partitions (full meaning closed for data) and one partition (current) that was filling with data

In the example above collection #1 and #2 would be closed for data as we are filling collection #3.  Once collection #3 is full then we drop collection #1, add collection #4 and make that the one that is accepting data.

Conclusion

This is simple to do when you know how but it does seem like a long winded approach.  I would like to see something a little less verbose.

Using Visual Studio to Work With Hive Scripts

I was building some HDInsight Azure Data Factory pipelines the other day and the Hive scripts were not doing what I wanted to them to do (although they were doing what was being asked).I needed to isolate the Hive scripts to see what they were doing at each stage of the script. I didn’t want to have to invoke the whole ADF just to look at the Hive section. I was in Visual Studio already so troubleshooting was easy from here. Here’s how I solved it.

Open up Server Explorer

Connect to your Azure Subscription

Navigate to the HDInsight node

Expand your clusters

Right click on the cluster of choice and select “Write a Hive Query”

Paste/Write your query

Another quirk of my Hive Script was that I was passing in values from the activity to the Hive Script itself. I knew what the values were that I was using but didn’t want to change the script to use hard coded values instead of the reference to the config. Here’s what I mean

I wanted to pass in my values as parameters to the script in Visual Studio. No problem. Once you have finished writing your query

Hit Submit | Advanced

Now you can enter your configuration key value pairs

This for me was a really quick and useful way of troubleshooting my Hive scripts. Hope you find it useful.

Azure Streaming Analytics Lag() Function with example

The Lag() function in Azure Streaming Analytics is documented here.  I have been asked a few times now about Lag() and if it can refer to the projected values of another Lag() function in the previous event.  Let’s see.

 

The Event

My events are really simple and their schema is:

 

{
type: "string",
readingtaken: "datetime",
reading: 0
}

 

The Query

 

My query in ASA is

 

select
type,
readingtaken,
case
when reading IS NULL then LAG(reading,1,0) OVER (PARTITION BY type LIMIT DURATION(hh,24))
else reading
end as reading
into
dest
from
readings
where
type = 'temperature'

 

The Data Flow

 

Here are my events as they flow through the stream (these are the raw events and not what is projected rom the ASA query)

 

 

 

Here is what you as the consumer will see i.e. the projected events.

 

 

 

The Explanation

 

The event that is of interest is E3.

 

Here is an explanation of the projected events

 

E1 has the reading attribute and value as does E1(Projected)

E2 does not have the attribute and value but E2(Projected) uses the value from E1.

E3 again does not have the attribute and value.  E3 looks to E2 and not E2(Projected).  In E2 the value of reading is NULL and as per the documentation this is used.

 

 

Summary

Using Lag() you cannot or rather do not refer to the projected value of the previous event.  You refer to the actual event itself.  In the example above the reading attribute had an explicit value of NULL in E2 so that is what was used in E3

Config Files in Azure Data Factory

As you know I have started to use Visual Studio to publish my ADF solutions to the Azure cloud. I like it for ease of use and integration with TFS. I came across a piece of functionality that I had not seen before but currently is really only half baked (it works fine but the implementation needs work).
The feature is configs. On your project right click and choose Add | New Item.

The properties are easy to specify and I would imagine the most often specified property will be the connectionString for a linked service. Like this

OK so that part is easy and makes sense. The next part is where the dots are not joined up yet. As you can see from my project below I have two config files. One for Dev and one for UAT

When I build the project in the Release/Debug folder it creates sub folders for each of the configurations. The configuration parameters have been applied to my JSON objects. I can then take these files from the relevant folder and deploy them to the correct ADF using Powershell or the Portal

This seems very long winded and not really a good experience. Another way to do it is:

Right Click on your project | Unload Project

Right Click Project | Edit xxxx.dfproj

Find the section ADFConfigFileToPublish

Insert your config name (I.e.DevConfig.json)

Save | reload the project

In the future I fully expect a better “deployment with config” experience.

Looking at Azure Data Factory in VS2013

I have been using Azure Data Factory now for quite a while.  When I first saw it, it was difficult to maintain and code.  The best I could say about it at that point was that it “has promise”.  Well I am happy to report that using it today is definitely a whole load easier.  This post is a very quick look at how designing your ADF flows has become clean and intuitive.  I am particularly pleased to see the addition of a VS plugin.  In order to use Visual Studio you will currently need VS2013 and the Azure SDK for .Net 2.7.  Once installed you will be able to create ADF solutions

 

Creating an empty solution

 

 

 

Now you can go ahead and create your entities.

 

Linked Services

 

 

Tables

 

 

Pipelines

 

 

Now the thing about the pipeline is that it is by far the most complicated entity in ADF.  What I really like is that the ADF team have given us a head start in getting the syntax right by creating placeholders for us.  Below I have added a Hive pipeline to my project.

 

 

 

Summary

 

Azure Data Factory is a great PaaS offering and now that you can design your solutions in Visual Studio I think the take up of it will increase dramatically.