Category Archives: Map/Reduce

Ok, so if you’ve been following along, you’ll already know:

1. Installed RabbitMQ, and
2. Written a Twitter OnRamp that subscribes to a sample Twitter feed and publishes it on RabbitMQ, and
3. Written a RuleBot that takes the tweet and breaks it into chunks, putting the chunks back onto the bus.

And now I’m going to show the Esper code that gives us the top URL’s in the last X seconds found in the Twitter feed.  Here’s a screenshot.

.

Here’s some output in text mode (easier to read):

Event received: {URL=http:\\cnn.com, Total=731}

Event received: {URL=http:\\colinclarkeventprocessing.com, Total=643}

Event received: {URL=http:\\rabbitmqm.com, Total=636}

Event received: {URL=http:\\cnn.com, Total=738}

Event received: {URL=http:\\colinclarkeventprocessing.com, Total=651}

Event received: {URL=http:\\rabbitmqm.com, Total=642}

Event received: {URL=http:\\cnn.com, Total=742}

Event received: {URL=http:\\colinclarkeventprocessing.com, Total=660}

Event received: {URL=http:\\rabbitmqm.com, Total=643}

Event received: {URL=http:\\cnn.com, Total=743}

Event received: {URL=http:\\colinclarkeventprocessing.com, Total=660}

Event received: {URL=http:\\rabbitmqm.com, Total=652}

Event received: {URL=http:\\cnn.com, Total=752}

Event received: {URL=http:\\colinclarkeventprocessing.com, Total=661}

Event received: {URL=http:\\rabbitmqm.com, Total=654}

Event received: {URL=http:\\cnn.com, Total=759}

Essentially, what’s happening here is that we’re:

1. Getting demo data (so I know what the relative ratio’s are to get a feel for whether or not this is correct), and
2. Pumping it into Esper via an URL object, and
3. Executing a statement (“select URL, sum(count) as Total from TwitterURL.win:time(5 seconds) group by URL”).

This statement gives us the total number of times an URL has been tweeted in the last 5 seconds.  Whenever that URL is tweeted, the sum is calculated, and a listener is called.  This is a very straightforward statement – we could tweak the statement a bit more, or do some stuff in code-like creating a map of URL’s with the current sum (rank).  Or maybe even another stream created.  Either way would expose the information (most popular URL’s) to any interested party.

All that’s left now is to get all of this up and running in the cloud – and that’s what I call real CEP – Cloud Event Processing.  Check in next week.

It’s time to delve a little deeper into what’s proving to be a thorny issue for doing anything in the cloud – especially if you’ve got concerns over once, only once, and in sequence – something that Financial Market’s participants have to deal with every day. We’re not going to get into that level of depth for this post. But I am going to show one way to deploy CEP in the cloud and use a CEP based continuous query to provide load balancing in our classic Map/Reduce (or what has been referred to as a streaming fold) example.

Load Balancing – Partitioning

In our example, we could instruct the RuleBots to only listen to tweets beginning from A-D, E-G, etc. The CEP Service Manager would start up how ever many instances initially required and set up this initial subscription. In the event that the subscriptions needed to change for load balancing requirements, the Load Balancer could stop all of the instances in the Service Pool, re-partition them, and then start them up. All without losing any messages and without processing any message more than once. Even without dealing with in-order (sequence based) processing, one can see how this particular approach leaves a little bit to be desired.

Load Balancing – Selecting a Destination

In the example below, we can see the RuleBots sending resource utilization statistics to the CEP Load Balancing Process.  In this process, the CEP Load Balancing Process listens to all relevant processes comprising a CEP Service Pool and constructs what I call a validity stream – some others call it an ‘update in place’ query.  Essentially, selecting from the stream (“select virtual-node from cep-service-pool where cep-service = “WORDCOUNT” order by resource-utilization desc” ) gives the OnRamp a destination for the current tweet JSON object.  I like this approach a lot.  New processing resources can be added to a CEP service pool dynamically, the CEP load balancer doesn’t really need to know anything about them – the service just publishes their stats and they’re eligible for processing.  Cool.  A picture paints the picture below.