Scaling Azure Event Hubs Processing with Worker Roles

This post will show how to host an Azure Event Hubs EventProcessorHost in a cloud service worker role.

Background

Imagine that you have a scenario where you have a solution that needs to ingest millions of events per second, and you need to process these events in near real time.  There are many scenarios where this might occur.  Perhaps you built a solution for a consumer device such as Microsoft Band using the recently announced Microsoft Band SDK preview, you have a popular game where you are processing telemetry events, or you might have an IoT solution where you are processing events from many devices.  Architecting for this type of solution poses many challenges.  How do you enable ingest of messages at this scale?  More importantly, how do you enable processing of the messages in a manner which does not create a performance bottleneck?  How do you independently scale the message ingest and message processing capabilities?  Most importantly, how do you achieve this for very little cost?

Event Hubs and EventProcessorHost

Azure Event Hubs enables massive scale telemetry processing for solutions that require logging millions of events per second in near real time.  It does this by creating a gigantic write-ahead stream where the ingest and processing of data is handled separately.  One way to think of this is as a massively scalable queue implementation.  However, unlike queues, there is no concept of queue length in Azure Event Hubs because the data is processed as a stream.  I wrote a demonstration of the sender side in the post Use JMS and Azure Event Hubs with Eclipse, where many senders send information to an Event Hub named “devicereadings”.  The Event Hub then distributes the messages across partitions.

image

The question is now how to process the events in a scalable manner.  That’s exactly what Event Hubs enables, scalable event processing.  The Azure Event Hubs team made this processing easy by introducing the EventProcessorHost that enables a partitioned consumer pattern.  Consider the following scenario where I have a worker role that is processing all of the stream data from the Event Hub.  The EventProcessorHost will automatically create an instance for each partition. 

image

We might see that a single worker role is not sufficient for our solution, that the worker role becomes CPU bound.  As we saw in my previous post, Autoscaling Azure–Cloud Services, scaling a worker role creates a new instance of the worker role.  The benefit of using the EventProcessorHost is that we can add more worker role instances, and the partitions will be balanced across them.  We don’t have to manage the number of processor instances, the EventProcessorHost handles that for us.

image

Using the EventProcessorHost is simple because you need to implement one interface, IEventProcessor, the bulk of the work will be in the ProcessEventsAsync class.

image

Let’s work through building the solution. 

Show Me Some Code

I am going to use the Service Bus Event Hubs Getting Started sample as the basis for this post, I made some changes to the code and I’ll explain along the way.  First, I created a class library called EventHubDemo.Common.  It contains NuGet references to Json.NET, Microsoft Azure Service Bus, and Windows Azure Configuration Manager.

image

It contains two classes:  MetricEvent.cs and EventHubManager.cs.

image

MetricEvent is just a class used to (de)serialize JSON data.

MetricEvent.cs

  1. using System.Runtime.Serialization;
  2.  
  3. namespace EventHubDemo.Common.Contracts
  4. {
  5.     [DataContract]
  6.     public class MetricEvent
  7.     {
  8.         [DataMember]
  9.         public int DeviceId { get; set; }
  10.         [DataMember]
  11.         public int Temperature { get; set; }
  12.     }
  13. }

The next class, EventHubManager.cs, is just a helper class for working with Event Hubs.  It centralizes common logic that is used across multiple projects to work with the Event Hub connection string and to create an Event Hub.

Code Snippet

  1. using Microsoft.ServiceBus;
  2. using Microsoft.ServiceBus.Messaging;
  3. using System;
  4. using System.Diagnostics;
  5.  
  6. namespace EventHubDemo.Common.Utility
  7. {
  8.     public class EventHubManager
  9.     {
  10.         public static string GetServiceBusConnectionString()
  11.         {
  12.             string connectionString = Microsoft.WindowsAzure.CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");
  13.             
  14.             if (string.IsNullOrEmpty(connectionString))
  15.             {
  16.                 Trace.WriteLine("Did not find Service Bus connections string in appsettings (app.config)");
  17.                 return string.Empty;
  18.             }
  19.             ServiceBusConnectionStringBuilder builder = new ServiceBusConnectionStringBuilder(connectionString);
  20.             builder.TransportType = TransportType.Amqp;
  21.             return builder.ToString();
  22.         }
  23.  
  24.         public static NamespaceManager GetNamespaceManager()
  25.         {
  26.             return NamespaceManager.CreateFromConnectionString(GetServiceBusConnectionString());
  27.         }
  28.  
  29.         public static NamespaceManager GetNamespaceManager(string connectionString)
  30.         {
  31.             return NamespaceManager.CreateFromConnectionString(connectionString);
  32.         }
  33.  
  34.  
  35.         public static void CreateEventHubIfNotExists(string eventHubName, int numberOfPartitions, NamespaceManager manager)
  36.         {
  37.             try
  38.             {
  39.                 // Create the Event Hub
  40.                 Trace.WriteLine("Creating Event Hub...");
  41.                 EventHubDescription ehd = new EventHubDescription(eventHubName);
  42.                 ehd.PartitionCount = numberOfPartitions;
  43.                 manager.CreateEventHubIfNotExistsAsync(ehd).Wait();
  44.             }
  45.             catch (AggregateException agexp)
  46.             {
  47.                 Trace.WriteLine(agexp.Flatten());
  48.             }
  49.         }
  50.  
  51.     }
  52. }

This makes it easy to reuse the logic in a Console application that sends messages or in an Azure worker role that receives them.

EventProcessorHost in an Azure Worker Role

The next step is to create a Cloud Service.  I create a new Azure Cloud Service project named “EventProcessor” in Visual Studio.

image

Click OK, and you are prompted for the type of role you want to create.  Choose a worker role, and rename it to something like “ReceiverRole”.

image

Click OK, and you now have two projects: the worker role code and the deployment project.

image

The bulk of the work will be in the ReceiverRole project.  Add a reference to the EventHubDemo.Common library that we created previously.  Next add the NuGet package “Microsoft.Azure.ServiceBus.EventProcessorHost”, which will add the required dependencies.

image

Next we add a class called SimpleEventProcessor.cs.  This class will implement the IEventProcessor interface that we mentioned previously.  When the OpenAsync method is called by the EventProcessorHost, we will write out the partition that it corresponds to.  When a message is received in the ProcessEventsAsync method, we write the message out to Trace output.  When the CloseAsync method is called, we write that out to trace as well.

SimpleEventProcessor.cs

  1. using EventHubDemo.Common.Contracts;
  2. using Microsoft.ServiceBus.Messaging;
  3. using Newtonsoft.Json;
  4. using System;
  5. using System.Collections.Generic;
  6. using System.Diagnostics;
  7. using System.Linq;
  8. using System.Text;
  9. using System.Threading.Tasks;
  10.  
  11. namespace ReceiverRole
  12. {
  13.     class SimpleEventProcessor : IEventProcessor
  14.     {
  15.         PartitionContext partitionContext;
  16.  
  17.         public Task OpenAsync(PartitionContext context)
  18.         {
  19.             Trace.TraceInformation(string.Format("SimpleEventProcessor OpenAsync.  Partition: '{0}', Offset: '{1}'", context.Lease.PartitionId, context.Lease.Offset));
  20.             this.partitionContext = context;
  21.             return Task.FromResult<object>(null);
  22.         }
  23.  
  24.         public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> events)
  25.         {
  26.             try
  27.             {
  28.                 foreach (EventData eventData in events)
  29.                 {
  30.                     try
  31.                     {
  32.                         var newData = this.DeserializeEventData(eventData);
  33.  
  34.                         Trace.TraceInformation(string.Format("Message received.  Partition: '{0}', Device: '{1}', Data: '{2}'",
  35.                             this.partitionContext.Lease.PartitionId, newData.DeviceId, newData.Temperature));
  36.                     }
  37.                     catch (Exception oops)
  38.                     {
  39.                         Trace.TraceError(oops.Message);
  40.                     }
  41.                 }
  42.  
  43.                 await context.CheckpointAsync();
  44.  
  45.             }
  46.             catch (Exception exp)
  47.             {
  48.                 Trace.TraceError("Error in processing: " + exp.Message);
  49.             }
  50.         }
  51.  
  52.         public async Task CloseAsync(PartitionContext context, CloseReason reason)
  53.         {
  54.             Trace.TraceWarning(string.Format("SimpleEventProcessor CloseAsync.  Partition '{0}', Reason: '{1}'.", this.partitionContext.Lease.PartitionId, reason.ToString()));
  55.             if (reason == CloseReason.Shutdown)
  56.             {
  57.                 await context.CheckpointAsync();
  58.             }
  59.         }
  60.  
  61.         MetricEvent DeserializeEventData(EventData eventData)
  62.         {
  63.  
  64.             string data = Encoding.UTF8.GetString(eventData.GetBytes());
  65.             return JsonConvert.DeserializeObject<MetricEvent>(data);
  66.         }
  67.     }
  68. }

Easy enough so far.  Now add a class, Receiver.cs.  This class will encapsulate the logic for registering and unregistering an IEventProcessor implementation.

Receiver.cs

  1. using Microsoft.ServiceBus.Messaging;
  2. using System.Diagnostics;
  3.  
  4. namespace ReceiverRole
  5. {
  6.     class Receiver
  7.     {        
  8.         string eventHubName;
  9.         string eventHubConnectionString;
  10.         EventProcessorHost eventProcessorHost;
  11.         
  12.         public Receiver(string eventHubName, string eventHubConnectionString)
  13.         {
  14.             this.eventHubConnectionString = eventHubConnectionString;
  15.             this.eventHubName = eventHubName;
  16.         }
  17.  
  18.         public void RegisterEventProcessor(ConsumerGroupDescription group, string blobConnectionString, string hostName)
  19.         {
  20.             EventHubClient eventHubClient = EventHubClient.CreateFromConnectionString(eventHubConnectionString, this.eventHubName);
  21.  
  22.             if (null == group)
  23.             {
  24.                 //Use default consumer group
  25.                 EventHubConsumerGroup defaultConsumerGroup = eventHubClient.GetDefaultConsumerGroup();
  26.                 eventProcessorHost = new EventProcessorHost(hostName, eventHubClient.Path, defaultConsumerGroup.GroupName, this.eventHubConnectionString, blobConnectionString);
  27.             }
  28.             else
  29.             {
  30.                 //Use custom consumer group
  31.                 eventProcessorHost = new EventProcessorHost(hostName, eventHubClient.Path, group.Name, this.eventHubConnectionString, blobConnectionString);
  32.             }
  33.  
  34.             
  35.             Trace.TraceInformation("Registering event processor");
  36.             
  37.             eventProcessorHost.RegisterEventProcessorAsync<SimpleEventProcessor>().Wait();
  38.         }
  39.  
  40.         public void UnregisterEventProcessor()
  41.         {
  42.             eventProcessorHost.UnregisterEventProcessorAsync().Wait();
  43.         }
  44.     }
  45. }

The final bit is to host this in an Azure cloud service using a worker role.  I updated the WorkerRole.cs class.

Code Snippet

  1. using EventHubDemo.Common.Utility;
  2. using Microsoft.WindowsAzure;
  3. using Microsoft.WindowsAzure.ServiceRuntime;
  4. using System;
  5. using System.Diagnostics;
  6. using System.Net;
  7. using System.Threading;
  8.  
  9. namespace ReceiverRole
  10. {
  11.     public class WorkerRole : RoleEntryPoint
  12.     {                
  13.         private Receiver receiver;        
  14.         private readonly ManualResetEvent runCompleteEvent = new ManualResetEvent(false);
  15.  
  16.         public override void Run()
  17.         {
  18.             Trace.TraceInformation("ReceiverRole is running");
  19.  
  20.             //Get settings from configuration
  21.             var eventHubName = CloudConfigurationManager.GetSetting("eventHubName");
  22.             var consumerGroupName = CloudConfigurationManager.GetSetting("consumerGroupName");
  23.             var numberOfPartitions = int.Parse(CloudConfigurationManager.GetSetting("numberOfPartitions"));
  24.             var blobConnectionString = CloudConfigurationManager.GetSetting("AzureStorageConnectionString"); // Required for checkpoint/state
  25.  
  26.             //Get AMQP connection string
  27.             var connectionString = EventHubManager.GetServiceBusConnectionString();
  28.  
  29.             //Create event hub if it does not exist
  30.             var namespaceManager = EventHubManager.GetNamespaceManager(connectionString);
  31.             EventHubManager.CreateEventHubIfNotExists(eventHubName, numberOfPartitions, namespaceManager);
  32.             
  33.             //Create consumer group if it does not exist
  34.             var group = namespaceManager.CreateConsumerGroupIfNotExists(eventHubName, consumerGroupName);
  35.  
  36.             //Start processing messages
  37.             receiver = new Receiver(eventHubName, connectionString);
  38.  
  39.             //Get host name of worker role instance.  This is used for each environment to obtain
  40.             //a lease, and to renew the same lease in case of a restart.
  41.             string hostName = RoleEnvironment.CurrentRoleInstance.Id;
  42.             receiver.RegisterEventProcessor(group, blobConnectionString, hostName);
  43.  
  44.             //Wait for shutdown to be called, else the role will recycle
  45.             this.runCompleteEvent.WaitOne();
  46.         }
  47.  
  48.         public override bool OnStart()
  49.         {
  50.             // Set the maximum number of concurrent connections
  51.             ServicePointManager.DefaultConnectionLimit = 12;
  52.  
  53.             // For information on handling configuration changes
  54.             // see the MSDN topic at https://go.microsoft.com/fwlink/?LinkId=166357.
  55.             
  56.             bool result = base.OnStart();
  57.  
  58.             Trace.TraceInformation("ReceiverRole has been started");
  59.  
  60.             return result;
  61.         }
  62.  
  63.         public override void OnStop()
  64.         {
  65.             Trace.TraceInformation("ReceiverRole is stopping");
  66.  
  67.             this.runCompleteEvent.Set();
  68.             try
  69.             {
  70.                 //Unregister the event processor so other instances
  71.                 //  will handle the partitions
  72.                 receiver.UnregisterEventProcessor();
  73.             }
  74.             catch(Exception oops)
  75.             {
  76.                 Trace.TraceError(oops.Message);
  77.             }
  78.             
  79.             base.OnStop();
  80.  
  81.             Trace.TraceInformation("ReceiverRole has stopped");
  82.         }
  83.         
  84.     }
  85. }

When the worker role starts, we capture the trace output.  The worker role then starts to run, and we get settings from configuration in order to connect to the Event Hub and register our event processor.  A key point to call out is line 41 where we use the host name of the worker role instance, this is needed in order to obtain a lease for the partitions across multiple instances.  Also notice line 45 where we wait for a signal.  That signal is set on line 67 when the worker role is stopped.  The processing of events is handled on a separate thread, we only need to use the ManualResetEvent to prevent the Run method from completing (which would cause the worker role to recycle).

The sample code uses CloudConfigurationManager to obtain settings.  Those settings are part of the deployment process, shown in the next section.

Role Configuration

Go to the EventProcessor project that defines the deployment configuration for the worker role.  Expand the Roles folder and double-click on the ReceiverRole node.

image

When you double-click, go to the Settings tab.  This is where we provide the various configuration values.

image

Note: It is a best practice to avoid putting secrets such as connection strings in source code. You would typically set these as part of a deployment script. For more information, see Building Real-World Cloud Apps with Azure.  

You can also configure diagnostics for the worker roles.  We will use the diagnostics capabilities when we test out the solution.  On the Configuration tab, click the Configure button.

image

That allows you to configure the log level for the application.  I set it to Verbose.

image

Now right-click the deployment project and choose Publish.  You are prompted to create a cloud service and storage account.

image

Once created, go through the wizard to complete publishing.

image

Watch the progress in the Microsoft Azure Activity Log.

image

Once deployed, you will have a single production instance for your worker role.

Testing It Out

Once the role is deployed, go to Visual Studio and right-click the ReceiverRole node under Cloud Services and choose “View Diagnostics Data”.

image

On that page, expand the Microsoft Azure application logs section, then click View all data.

image

You will then see the Azure Storage table called WADLogsTable, and you can see the events that happened.  When the worker role is running, we see the entry “ReceiverRole is running”.  The SimpleEventProcessor is registered once, and the EventProcessorHost takes care of creating 16 instances, one for every partition (not all log entries are shown here, but there are 16 in the log). 

image

Let’s scale our service.  While we could use the autoscale service to scale by CPU, we can see how things work by manually scaling as well.  Let’s manually scale by going to the Azure management portal (https://manage.windowsazure.com) and setting the scale for the cloud service to 4.

image

Save, and wait while the new instances are created and our packages are deployed automatically to the new instances.

image

Once the 3 new virtual machines are created, we go back to see what happened in the logs.  This is really cool to see how the EventProcessorHost automatically balances the processing across the available instances.  You can see here that partitions 1,10, and 12 are handled by instance 2, while partitions 3,13, and 15 are handled by instance 1.

image

Let’s scale the number of instances down to 2. 

image

This will cause our OnStop() method to be called in each worker role that is no longer needed, and our code to unregister the event processor is called.  This balances the processing back to the remaining worker role instances.  First we see the close operations:

image

Next we see that the closed partitions are opened using another available processor:

image

Testing Out the Sender

To test things out, I can use the same client that I built in the post Use JMS and Azure Event Hubs with Eclipse.  This is a simple web page that sends messages to the Event Hub using AMQP 1.0.  Alternatively, I can modify the code in the Service Bus Event Hubs Getting Started sample to send messages.  Once I send messages to the Event Hub, we see them processed in the event log, automatically balanced across partitions among the 2 available worker roles.

image

Summary

Event Hubs make it very easy to build highly scalable solutions that can process millions of messages per second in near real time.  This post shows how you can use an Azure worker role with EventProcessorHost to process the messages and how the load will automatically balance across the available processors.

I am not providing the sample code as a download because it is already provided in the post Service Bus Event Hubs Getting Started, and I have listed the code that I used in this post. 

For More Information

Use JMS and Azure Event Hubs with Eclipse

Building Real-World Cloud Apps with Azure – Free eBook!

Service Bus Event Hubs Getting Started – sample used as the basis of this post