Nati Shalom's Blog

Real-Time Big Data with Storm, Cassandra, and XAP

Real-time Big Data is becoming more and more common in Big Data systems.

One of the most common frameworks used for running real-time Big Data system is Twitter Storm backed by a NoSQL database, as shown in the diagram below:

 

In this architecture, Storm is used for processing data as it comes in and the NoSQL data store is used as the output for that processing as well as for reference data storage.

Challenges

This architecture presents a couple of challenges:

• Performance - Storm runs in-memory, and is therefore set to process large volumes of data at in-memory speed. However, a typical storm architecture needs to interface with a data source for its input and a data source for its output. In this context, the overall performance is now determined not by Storm itself but by the data input and output sources. Quite often these interfaces rely on file-based message queues for streaming, and NoSQL for data storage. These interfaces are at least an order of magnitude slower than Storm itself, and therefore become the limiting factor.
• Complexity - Storm itself consists of several moving parts, such as a coordinator  (ZooKeeper), state manager (Nimbus), and processing nodes (Supervisor). The NoSQL data store also comes with its own cluster management. In addition, a typical Big Data system comes with more components such as the application front end, a reporting database, and more. This makes the process of managing the deployment, configuration, fail-over and scaling of such a system quite complex.

Meeting the Performance Challenge

Given that Storm itself runs in memory, it only makes sense that in order for Storm to run at maximum capacity the streaming and data store interfaces should be implemented in-memory as well, as shown below:

As you can see in the above architecture, we added two interfaces that rely on a built-in Storm plug-in - one for the data inputs and the second for data output. In both cases, the underlying implementation is memory-based, and thus removing the impedance that the previous architecture included.

You can read the full details on how this integration works, including a code example, in DeWayne Filppi's post on integration of in-memory computing with Cassandra.

The second optimization that can be added is to the NoSQL data store.

As most analytics applications tend to be read-mostly, we can speed up access to the NoSQL data store using an in-memory local cache. This architecture speeds up read performance between 10 to 1000 times as outlined by Shay Hassidim in his post on real-time Big Data performance.

Meeting the Complexity Challenge

To meet the complexity challenge, we will use a DevOps automation approach using Cloudify in conjunction with Chef or Puppet.

With this approach, we wrap every service with a deployment recipe that abstracts the underlying details on how to manage Storm, Cassandra, and our in-memory data store. Cloudify uses these recipes to automate the deployment of the entire stack. In this way, you only need to interact with Cloudify for the deployment, configuration, and also scaling and fail-over of your stack rather than separately manage each individual component. In addition, we use Cloudify to abstract the underlying infrastructure. This enables us to use the same deployment recipe across different environments. One for testing, another for production, etc. We can also use the same approach to deploy our apps based on the type of workload. We can use a public cloud for running sporadic workload, and thus leverage the elasticity of the cloud and also enable us to create and scale the environment as well as rip it off completely when done. At the same time, we can choose bare-metal machines for I/O intensive workloads, and so on.

You can read more on how you can set Storm using Cloudify on DeWayne's post on Storm and the cloud.

Cloudify comes with a rich set of built-in recipes for other Big Data services as well, making the integration process an out of the box experience. The main ones are listed below:

Final Notes - Optimizing without Code Change

Many real-time Big Data systems are now based on Twitter Storm and a NoSQL data store such as Cassandra.

In this post I tried to outline how we can optimize this architecture by addressing two areas: performance, and management.

The good news is that all this is possible to achieve seamlessly, without any code change. Let me explain:

In the case of Storm we used built-in plug-ins - Spout for streaming and Trident as the data store interface. In this way we can simply plug our memory-based plug-ins under these two integration points. All our existing Storm business logic would work pretty much the same.

We use the same plug-in approach to integrate our in-memory data store with our NoSQL data. This integration makes the data flow between our real time streaming to our NoSQL storage fairly transparently. In addition, it allows us to plug in different NoSQL data stores such as MongoDB, CouchBase, etc., giving us another degree of flexibility.

The same applies to our management layer. Existing Storm and NoSQL deployments are wrapped into plugged in recipes, and don't require any specific code changes.

Not Mutually Exclusive

We don't have to implement all of the optimizations to gain the benefit of this architecture. Each of the optimization points can be plugged in  independently of the others. For  example, if your main pain point is complexity, you can start by only by adding the DevOps automation first. At the same time if your main pain point is performance, you can use the memory-based plug-ins to speed up processing.

The other advantage of the combined architecture that I haven't discussed in this post is that it provides more flexibility in the degrees of consistency and availability in which you set your system, as I outlined in one of my recent posts on the subject In Memory Computing (Data Grid) for Big Data.

References:

• Big Data Real-Time Performance
• GigaSpaces and Storm Part 2: XAP Integration
• GigaSpaces And Storm Part 1: Storm Clouds
• In Memory Computing (Data Grid) for Big Data
• Check out how GigaSpaces XAP and Cassandra integration works

Making Hadoop Run Faster

One of the challenges in processing data is that the speed at which we can input data is quite often much faster than the speed at which we can process it. This problem becomes even more pronounced in the context of Big Data, where the volume of data keeps on growing, along with a corresponding need for more insights, and thus the need for more complex processing also increases.

Batch Processing to the Rescue

Hadoop was designed to deal with this challenge in the following ways:

1. Use a distributed file system: This enables us to spread the load and grow our system as needed.

2. Optimize for write speed: To enable fast writes the Hadoop architecture was designed so that writes are first logged, and then processed. This enables fairly fast write speeds.

3. Use batch processing (Map/Reduce) to balance the speed for the data feeds with the processing speed.

Batch Processing Challenges

The challenge with batch-processing is that it assumes that the feeds come in bursts. If our data feeds come in on a continuous basis, the entire assumption and architecture behind batch processing starts to break down.

If we increase the batch window, the result is higher latency between the time the data comes in until the time we actually get it into our reports and insights. Moreover, the number of hours is finite -- in many systems the batch window is done on a daily basis. Often, the assumption is that most of the processing can be done during off-peak hours. But as the volume gets bigger, the time it takes to process the data gets longer, until it reaches the limit of the hours in a day and then we face dealing with a continuously growing backlog. In addition, if we experience a failure during the processing we might not have enough time to re-process.

Speed Things Up Through Stream-Based Processing

The concept of stream-based processing is fairly simple. Instead of logging the data first and then processing it, we can process it as it comes in. 

A good analogy to explain the difference is a manufacturing pipeline. Think about a car manufacturing pipeline: Compare the process of first putting all the parts together and then assembling them piece by piece, versus a process in which you package each unit at the manufacturer and only send the pre-packaged parts to the assembly line. Which method is faster?

Data processing is just like any pipeline. Putting stream-based processing at the front is analogous to pre-packaging our parts before  they get to the assembly line, which is in our case is the Hadoop batch processing system.

As in manufacturing, even if we pre-package the parts at the manufacturer we still need an assembly line to put all the parts together. In the same way, stream-based processing is not meant to replace our Hadoop system, but rather to reduce the amount of work that the system needs to deal with, and to make the work that does go into the Hadoop process easier, and thus faster, to process.

In-memory stream processing can make a good stream processing system, as Curt Monash’s points out on his research traditional databases will eventually end up in RAM. An example of how this can work in the context of real-time analytics for Big Data is provided in this case study, where we demonstrate the processing of Twitter feeds using stream-based processing that then feeds a Big Data database for the serving providing the historical agregated view as described in the diagram below.

Faster Processing the Google Way: Using Stream-Based Processing Instead of Map/Reduce

Due to a lack of alternatives at the time, in many Big Data systems today Map/Reduce is used in areas where it wasn't a very good fit in the first place.  A good example is using Map/Reduce for maintaining a global search index. With Map/Reduce, we basically rebuild the index, where it would actually make more sense to update it with changes as they come in.

Google moved large part of its index processing from Map/Reduce into a more real-time processing model, as noted in this recent post:

So, how does Google manage to make its search results increasingly real-time? By displacing GMR in favor of an incremental processing engine called Percolator. By dealing only with new, modified, or deleted documents and using secondary indices to efficiently catalog and query the resulting output, Google was able to dramatically decrease the time to value. As the authors of the Percolator paper write, ”[C]onverting the indexing system to an incremental system … reduced the average document processing latency by a factor of 100.” This means that new content on the Web could be indexed 100 times faster than possible using the MapReduce system!

..Some datasets simply never stop growing ..it is why trigger-based processing is now available in HBase, and it is a primary reason that Twitter Storm is gaining momentum for real-time processing of stream data.

Final Notes

We can make our Hadoop system run faster by pre-processing some of the work before it gets into our Hadoop system. We can also move the types of workload for which batch processing isn't a good fit out of the Hadoop Map/Reduce system and use Stream Processing, as Google did.

Interestingly enough, I recently found out that Twitter Storm came up with an option to integrate an in-memory data store into Storm through the Trident-State project. The combination of the two makes lots of sense and something were currently looking at right now so stay tuned.

 

References

• Build Your Own Twitter Like Real Time Analytics - A Step By Step Guide (Blog Post)
• Build Your Own Twitter Like Real Time Analytics - A Step By Step Guide (Online Tutorial)
• Real Time analytics for Big Data: Facebook's New Realtime Analytics System
• Beyond hadoop: fast queries from big data
• Traditional databases will eventually wind up in RAM
• Why the days are numbered for Hadoop as we know it

Related articles

Elephants and Eyeballs: Real-Time Big Data in the Real World

Cassandra on ACID

GigaOm: 'Hadoop's days are numbered...' Are they?

Hadoop Officially a Big Deal for Big Data

Big Data In the Cloud Using Cloudify

Edd Dumbill wrote an interesting article on O’Reilly Radar covering the current solutions for running Big Data in the Cloud

Big data and cloud technology go hand-in-hand. Big data needs clusters of servers for processing, which clouds can readily provide.

Edd touched briefly on the role of PaaS for delivering Big Data applications in the cloud

Beyond IaaS, several cloud services provide application layer support for big data work. Sometimes referred to as managed solutions, or platform as a service (PaaS), these services remove the need to configure or scale things such as databases or MapReduce, reducing your workload and maintenance burden. Additionally, PaaS providers can realize great efficiencies by hosting at the application level, and pass those savings on to the customer.

Even though Edd’s article covers all the different forms of running Big Data on private and public clouds, the article focuses mainly on the public cloud offering from Amazon, Microsoft and Google.

In this post, I wanted to cover more specifically how I see the evolution of cloud application platforms (PaaS) to support Big Data. I’ll refer specifically to Cloudify which was designed primarily to support Big Data applications. 

 

 

Big Data in the cloud using Cloudify

Background

Most of the PaaS solutions out there started by focusing on simple web application deployments on Ruby, Java and Node.js. Unlike other PaaS solutions, when we designed Cloudify we picked Big Data as the primary target for Cloudify, and started by supporting popular NoSQL clusters such as Cassandra and MongoDB, as well as providing the equivalent of Amazon RDS by providing recipes for MySQL. Our goal was to make Big Data deployments a first class citizen within Cloudify. To this end,when you download Cloudify you'll notice that ALL of our examples comes with pre-integrated Big Data deployments.

There are couple of reason that brought us to make that decision: 

• Managing large data clusters is a core expertise at GigaSpaces

Most people know GigaSpaces for our In-Memory Data Grid solution known as XAP (eXtreme Application Platform). Over the past 10 years, as our customer deployments grew substantially, we realized that developing strong automation and cluster management is as critical as handling data consistency, performance, and latency in of our data-grid product. In a large cluster, if something breaks it’s going to literally be impossible to handle that failure through manual procedures.  For that reason, we developed lots of IP around automation of data cluster deployment which resulted in a unique self-managed data cluster.

• Cloudify is a natural evolution of GigaSpaces Data Cluster

When we built Cloudify it made a lot of sense to take the IP that we developed for managing GigaSpaces cluster and simply generalize it so it would fit with any other framework. In this way, we could leverage the years of experience  as well as development in this area, and gain a significant head-start.

• Big Data applications are complex

Big Data applications tend to be fairly complex, which makes them an ideal candidate for the sort of automation and management that Cloudify can offer.

• Big Data applications have a lot in common with XAP applications

Both need automation of data, failover and recovery, both fit into large cluster deployments, and both share similar partitioning and other clustering architecture. 

What makes Big Data platform different than any other application

Most of the existing orchestration systems were designed to handle stateless processes. Moving data is a completely different ballgame as you need to think of:

• Primary and Backup dependency 
• Availability - moving data without losing it.
• Moving processes to the data rather than the other way around.
• Data replication within and across sites 

Automating any of these processes through general orchestration tooling like Chef or Rightscale can become a fairly involved and complex process, with lots of pitfalls with handling edge scenarios for example the handshake process that is often involved when automating a data node failure, including a split-brain scenario.

In Cloudify we were able to curve out lots of that logic from the user, for example Cloudify will automatically ensure that primaries and backups won’t run on the same node or data center in case of disaster recovery. You don't need to do anything but tag your cluster nodes with a zone-tag. 

Managing Big Data applications != Managing Big Data storage

Managing data clusters is one thing. Being able to process the data is yet another challenge that we need to think about when we’re dealing with application platforms as I noted in one of my earlier post.

The main challenge is that quite often the management of the data processing logic is built on completely different scaling, availability and monitoring tools than the one used for managing our Big Data deployment. It turns out, that this silo thinking leads to a whole set of complexities starting from the inconsistency in having multiple managers, each determined in a different way when there is a failure or scaling event, and that quite often end up conflicting with one another. Having lots of moving parts is yet another challenge that makes the entire deployment pretty much a complete mess.

Built-in support for In Memory Stream based processing

In an earlier post, 5 Big Data predictions for 2012, Edd pointed out that streaming data processing is going to be one of the main trends for Big Data. 

Over the next few years, we'll see the adoption of scalable frameworks and platforms for handling streaming, or near real-time analysis and processing. In the same way that Hadoop has been borne out of large-scale web applications, these platforms will be driven by the needs of large-scale location-aware mobile, social and sensor use.

Being also part of XAP, Cloudify already comes with built-in support for streaming Big Data processing. This means that building your own Facebook or Twitter-like real-time analytics can be as a simple as writing only small scripts that handle the analytics counters. All the rest, i.e. scalability, availability, automation, cloud portability, management and monitoring, is covered by Cloudify as noted in this and this use case. 

Examples for Big Data applications running on Cloudify 

In the list below, I tried to put together a couple of references and examples that will make it easier for you to get started. The first reference points to a simpler scenario that will allow you to use Cloudify to deploy your Big Data database as a service. The other three references are full-application stack deployments that include the data-processing and web-tier of applications managed together with the Big Data database. 

Running a Big Data 'Database as a Service'

Cloudify comes with built-in recipes for Cassandra and MongoDB, as well as Solr (popular search engine), which makes it easy to deploy these database clusters on your local machine, data center or private/public cloud through a single command. In this way, you can use Cloudify to automate the database. 

Spring Travel application with Cassandra

Demonstrating a deployment of a JAVA-based commerce application with Cassandra as the database

The example includes recipes that provision the Cassandra database, create a schema, load the data, then spawn a Tomcat container, automatically injecting the reference to Cassandra, that then shows the custom management and monitoring of the application - all through a single command. A recent videocast showing how the travel application works on HP OpenStack cloud is available here.

Pet Clinic example with MongoDB

The Pet Clinic example does pretty much the same thing only using a sharded MongoDB cluster.

Twitter Real Time analytics example for Big Data 

The Twitter example  shows how you can attach real time stream-based processing for handling real live Twitter feeds, and how you can manage both the stream processing cluster and the Big Data cluster using Cloudify and run it on any cloud. The entire source code for this example is provided on Github.

Give it a try

To try out any of the examples you'll need to download Cloudify (latest) or (stable) build. Cloudify comes with all first three examples as part of the distribution under the recipes or examples directory. To try out these examples simply follow the steps from the Cloudify Quick Start Guide.  

 References:

• Big Data in the Cloud
• 5 Big Data predictions for 2012
• Big Data Application Platform
• Analytics for Big Data – Venturing with the Twitter Use Case
• Cloudify Download  (latest), (stable) 

Realtime Analytics for Big Data: A Facebook Case Study

via www.youtube.com

2012 Cloud, PaaS, NoSQL Predictions

2011 is coming to its end and now is a good time to start planning for 2012. I thought that a good start would be too look at my 2011 predictions and if my previous (and first) attempt to predict someting in that turbulent environment held any water...so, here is a quick recap of 2011.

Recap of 2011

Private vs. Public Cloud - As I noted in my recent post Public vs. Private Clouds I felt that during 2011 the debate around public vs. public cloud would become less interesting, as most of the industry has started to accept the fact that there is a need for both environments, and the important issue would become how to make them work well together. The most interesting development in that regard was Rackspace’s recent announcement about their plan to support OpenStack based private clouds, which shows that even public cloud providers have fully embraced this idea.

OpenStack is evolving from a movement into a viable reality - the momentum around OpenStack has gone through ups and downs throughout the year as happens with every new technology. However looking back, it appears that 2011 was a fairly successful year for OpenStack with its first public cloud available already in the market.  Dell and HP have started to offer the OpenStack based cloud to their customers, as has Citrix.  Rackspace announced their plan to provide official support including for those who want to build their own OpenStack environment…that's quite big considering the short timeframe from when the technology was first introduced…still there is a long way to go but the future looks promising - check out this survey in that regard.

PaaS adoption has been happening at a slower pace than expected, despite the fact that the trend remains consistent.  For PaaS startups 2011 was a fairly significant year with the acquisition of Heroku by Salesforce.  Amazon Redhat and VMware joined the PaaS arena; Amazon with Elastic Beanstalk, Redhat with their OpenShift initiative, VMware with CloudFoundry, adding to its previous acquisition of SpringSource vFabric. This was a fairly significant year for us at GigaSpaces as we launched a new product in this same domain that aims to completely change the way PaaS is being taught today (stay tuned…).

Google App Engine have no doubt been the disappointment of the year by literally killing GAE as we knew it (amongst many other things) with their new pricing model. 

Big Data has gone real time.  Facebook made a big announcement on how they moved their batch-oriented analytics system to real time analytics (See my previous posts on this subject here and here).  Twitter announced a launch of a new Real Time Analytics dashboard; while both join Google and Yahoo who have already started to make this shift.  Google has also been transforming their web analytics framework into real time. As I noted in my 2011 predictions, the entire debate around NoSQL and SQL didn't make sense, and indeed we’ve seen quite a few announcements both from Cassandra and Couchbase on their support for SQL-like query support.

In Memory Data Grids have also taken a similar approach where, with GigaSpaces, we’ve launched our JPA support, other Data Grid implementations such as Infinispan and Gemfire seems to be heading in that same direction each adding different levels of SQL support. The interesting development in this regard is that we were able to prove that you could actually mix and match Document/Schemaless APIs with SQL APIs and have the flexibility to choose the right language for the job (See online demo Same Data Any API).

All in all I think that I came fairly close - don't you think…?

Ok that gives me enough confidence to try the same thing for 2012. 

2012 predictions

Cloud

iCloud everywhere - IMO the biggest shift in Cloud is the fact that it’s going to become pretty much invisible to many of the end users as new mobile devices, operating systems and applications start to be designed with cloud support in mind. Apple iCloud and DropBox mark the beginning of this trend. Using cloud for collaboration and synchronization is definitely a killer app for many of the consumer based apps. I expect that in 2012 we’re going to continue to see a big push of many SaaS-based offerings in that space toward rich client support that uses the cloud as a backend and leverages the power of the new generation of advanced mobile devices. The difference is that those clients won’t be just another frontend for the same web UI, but something that will run almost entirely on the mobile device and will use more generic cloud services for synchronization and collaboration. This will create the need for more generic cloud services such as database as a service and other middleware services that can interact directly from mobile applications.

Moving from Amazon-centric clouds to Cloud Mashups – In 2011 we started to see new kinds of clouds starting to pop up. Literally every hardware vendor (IBM, Dell, HP,..), telco (ATT, Verizon, KT), and software provider (Oracle, Microsoft) are either developing or already offer something in this space. Each one tries to maintain a unique position to compete with Amazon either through SLAs, locality, security, or being more open through the support of OpenStack. In 2012, this movement is going to become even stronger as many of the players that have been making the investment during 2011 will come out full speed ahead in 2012.

Microsoft finally gets it with Azure - Microsoft has been around for a while with Azure with somewhat marginal success mostly around its .NET user base, an approach that is too narrow a play when it comes to cloud.  Their cloud strategy is coming into focus with the offering of a more ubiquitous cloud supporting technologies that were previously unheard of on a MSFT cloud platform - such as Java, PHP and it wouldn't be too far to assume that they will be supporting Linux applications in the cloud as well. 

Cost-driven Application Management - One of the things that is still fairly hard to measure in the cloud is cost, and more specifically how each component of our application and architecture contribute to cost.  This is specifically true during current market conditions which are going to put even more pressure on cost savings. Cost-driven application design patterns will start to emerge, and will become an integral part of any design for cloud applications just as scalability and performance are today. A new form of Cost Driven Application Management (CDA) will start to emerge to provide better insight on how our application behaves from a cost analysis perspective - Newvem is a new startup in that space that already launched their private beta.

Mission Critical Apps move into the Cloud - As the industry matures there is no reason why we should draw the line for cloud adoption at simple apps. The challenge will be mostly around performance, latency, and ensuring continuous availability. A new class of middleware and application platforms that are designed specifically for cloud environments will become more popular to help in that transition. On the other hand, Java and JEE specifically will finally become more cloud ready as I noted in an earlier post - Java and the Center Stage.

Network Gets into Cloud API Stack - While compute and storage have become virtualized to fit into the cloud, we haven't seen much advancement on the network layer. Many of the networking providers are now launching APIs to enable better control over the cloud network. Alcatel recently announced an interesting cloud proposition in this domain specifically targeting telcos.  The idea is to use the network as a vehicle for making distributed data centers look like one big cloud, making it possible to better leverage existing assets and offer SLA driven compute resources based on latency, location etc. Other cloud providers are also starting to open their network APIs starting from the Load Balancer down to the core switch. This opens up a new set of opportunities for integrating these network APIs with the upper layer of the application stack.

More OpenStack Clouds - 2011 was the just the beginning of that trend, 2012 will see more public and private cloud providers offering support for OpenStack APIs with RackSpace, Dell, and HP already making public announcements in this area. The interesting question in this regard would be how Citrix will play out their CloudStack acquisition with its OpenStack strategy. 

 

PaaS

DevOps and PaaS Converge into App DevOps PaaS - One of the topics that drew a lot of my personal interest last year was the DevOps movement. For odd reasons, most of that movement was driven by Ops and less by Devs.  In 2012, we’ll see many of the DevOps tools such as Chef and Puppet integrated into application platforms making it easier to deploy complex applications onto the cloud. In the same way, we’re going to see more Application Platforms adopting the automation and recipe model from the DevOps world into the application platform. The latter have the potential to transform the opinionated PaaS offerings as we know them today, with Heroku and GAE leading that trend, into a more open PaaS offering that better fits into the way users develop apps today and giving more freedom to choose your own stack, cloud, and application blueprint.

Beyond Google App Engine, Heroku - Heroku established itself as a one of the early PaaS providers in the market and is now expanding their offering to Java. CloudFoundry, DotCloud and others are slightly different but still follow the same path.  In 2012 you should expect more choices for completely different PaaS platforms starting with JEE PaaS offerings from Redhat, IBM, and Oracle, to private PaaS offerings which essentially are frameworks to build your own PaaS, DevOps PaaS offerings (see note above), as well as vertical PaaS for specific industries. Magento announced their plan to provide PaaS for eComm and it wouldn’t be crazy to assume that others will follow that same path. 

BigData

Not only Hadoop Centric - During 2010 and to a lesser degree 2011 Big Data discussions were pretty much centered around Hadoop.  NoSQL solutions such as Cassandra and Mongo are gaining fast adoption mainly due to the operational and development complexity that comes with Hadoop. That movement is going to continue at an even greater pace, as Hadoop gets fragmented between many vendors and frameworks such as EMC, MapR, Cloudera, Yahoo, IBM each claiming to own their own Hadoop distro. With new funding in the hands of many of the NoSQL startups I’d expect to see more complete solution stacks targeted at Big Data.

In Memory Data-Grid and NoSQL Become Integrated - During the early days of the NoSQL movement it wasn't clear how the two technologies fit together. As I noted in my previous post here, it actually makes more sense to integrate the two technologies in a context of real time analytics for Big Data or real time data processing for Big Data.  Indeed during 2011 I started to see more case studies showing the use of the two technologies as with Facebook and Twitter. MemBase is also a good example for that approach with their announcment earlier this year about their integration of Memcached and CouchDB together into a single product. At GigaSpaces we added built-in integration for Cassandra and MongoDB, as noted here, and plan to invest more in that direction during 2012.

A New Class of Big Data Application Platforms will Address the Development and Operational Complexity of Big Data Applications - As Big Data application become more mainstream we start to hit the next level of complexity, development, and operational complexity. Clearly plugging NoSQL into your architecture may address your scalability requirements but at the same time it’s going to make your development and management experience more complex.  Not because the products themselves are complex, but mostly it’s because it is less obvious how to build and design the application around these new technologies. As in previous years, the goal of application platforms is to ease that task by putting together an integrated stack that makes it easier to develop Big Data applications as I noted in my post on Big Data Application Platforms.

References

• The Future of Cloud Computing: Industry Predictions for 2012 | Cloud Computing Journal
• NoSQL Consolidation: CouchOne and Membase Merge to Form Couchbase
• 10 Predictions About Cloud Computing
• 2011 Cloud, PaaS, NoSQL Predictions
• Public vs Private clouds (Again!)- it's not about the cost
• Introducing Magento Go and the Magento Go Developer Platform. The Next Evolution in eCommerce
• Real Time Analytics for Big Data: An Alternative Approach
• Big Data Application Platform
• Java and the Center Stage.
• Same Data Any API

Java and Center Stage: Meet us at JavaOne 2011

Next week, I’m going to be in JavaOne. Unlike last year, where I expressed lots of skepticism about the way Oracle drives the Java Community, it seems that this year things are starting to get back on course – not that the skepticism has vanished completely, but at least there is a stronger sentiment that things are starting to settle down and take a more positive course.

It seems that now Java is taking center stage in the Cloud world, with more application platforms such as SalesForce/Heroku, VMware, Redhat/OpenShift, Cloud.com, JClouds, and obviously GigaSpaces providing a rich set of offerings that are based on Java.

One of the most interesting things in my opinion is that the combination of Cloud and Java brings new opportunities that until recently were unheard of:

 

• Heroku, which was very much Ruby-centric, added support for Java.
• GigaSpaces recently announced our partnership with Microsoft to bring Java to Microsoft's Azure community.

What’s interesting in those two examples is that the Cloud forces us to break away from the language wall. It brings communities that were living in completely isolated silos much closer together.

This forces a very different way of thinking about Java - and about the next generation Java platform specifically.

The new platform should be:

• Open for other languages (through integration points like Protobuff, Thrift, REST, and more, as well as the ability to run those languages)
• Embrace NoSQL implementations as an integral part of the platform
• Add support for dynamic data structures such as JSON, and schema-free document models
• Bring Dev and Ops closer by providing better exposure of operational aspect as part of the development frameowork
• Make cloud deployment a first class citizen

It seems that Java EE 7 is taking the right path in that direction. From Java EE 7 and Cloud Computing:

• The Java EE platform architecture is taking into account the operational impact of the cloud, more specifically by defining new roles, such as a PaaS administrator.
• The Java EE platform may also establish a set of constraints for PaaS-specific features, such as multi-tenancy, that deployments may have to obey. Applications may also be able to identify themselves as designed for cloud environments.
• All resource manager-related APIs, such as JPA, JDBC and JMS, will be updated to enable multi-tenancy. The programming model may be refined as well by introducing connectionless versions of the major APIs.
• Java EE will define a descriptor for application metadata to enable developers to describe certain characteristics of their applications that are essential for the purpose of running them in a PaaS environment. These may include being multitenancy-enabled, resources sharing, quality of service information, dependencies between applications.

What This Means

Java is regaining some of its luster from the last few years. It was being perceived as vaguely pedestrian and ordinary, and languages like Ruby and Python were being seen as more dynamic environments for rapid and scalable deployments.

Now we're seeing Java as the lingua franca of enterprise development. With so many environments being heterogeneous, a homogenous platform becomes a very desirable resource, and the JVM is able to run many languages in an integrated environment: Java, Ruby, Python, Scala, and Groovy, to name only a few, and many of these can run as compiled code and as scripts.

Even beyond the multiple programming language support of the JVM, you have very broad support for almost any number of remote procedure call mechanism you can think of. The old J2EE approach, where you defined your requirement and there was a specific way to fulfill it (which meant that every architecture was more or less nudged to the middle), is no longer a real limit.

If you can do it, you can do it in Java, without Java forcing you to sacrifice in the process.

This is a very powerful concept.

It puts your organization back under your control. This is desirable because you know your product and architecture better than a community process does. You know whether you need to support REST here, and SOAP there; you know you need to use NoSQL in this area and MySQL in that area.

A Remaining Challenge

Java EE 7 may be becoming aware of the new cloud-based environment, but it's still only one aspect of application deployment and design. What it does not address is the entire application environment.

This is where applications such as Cloudify come in. Cloudify treats every artifact involved with your deployment as a managed element.

Consider: when you deploy an enterprise application, you're not just deploying some operational code. That code has external requirements, like a database (or NoSQL warehouse); it might also rely on an Apache-based load balancer, or perhaps another independently-deployed artifact.

That's quite a burden on operations, because all of these are separately deployed and managed.

Cloudify, however, centralizes the management of each element your application is composed of, over the entire application's lifecycle, and provides a bridge to whatever environment you wish to use.

For example, you could have a MySQL database, used by a Java application hosted by a single Tomcat instance, fronted by an Apache httpd server, hosted on an internal Linux server. Cloudify can deploy and monitor each of those elements: the MySQL database, the schema and initial data; Tomcat; the Java application; Apache httpd, and the server upon which it all runs.

Now let's look at a larger deployment: a set of MySQL servers, a cluster of GigaSpaces XAP nodes, an instance of GlassFish hosting some web services, six .Net-based web services, and five httpd servers being load-balanced, deployed on a platform such as Microsoft Azure. From Cloudify's perspective, the deployment and management process is the same. You have more artifacts to deploy, and you're telling Cloudify to deploy on a different platform... that's it.

Cloudify can make sure each resource is running optimally, as well. If you can measure an attribute - such as CPU load, or disk space, or free memory - you can tell Cloudify to assert that some headroom exists, and what to do if a metric is exceeded.

For example, if you have a CPU running at 100% for an extended period of time, Cloudify can automatically deploy another instance of the component consuming the CPU, and configure load balancing so your application is more scalable, without human intervention.

Likewise, if you have four servers running a given process, all at 0% CPU load, you can reduce the number of containers with that component so that you're not overallocating resources, saving energy and money.

The key thought here:

You know how to solve your problem. We can help.

We'll be on the floor at JavaOne 2011, in booth #5002. We'll be happy to show you both XAP, our industry-leading application platform, and Cloudify, our deployment solution for the cloud-enabled world.

 

 

(This post was co-written with Joseph Ottinger, who unfortunately won't be able to make it to JavaOne this year. But he says that the best people in GigaSpaces will be there, so you're not missing out!)

Big Data Application Platform

It's time to think of the architecture and application platforms surrounding "Big Data" databases. Big Data is often centered around new database technologies mostly from the emerging NoSQL world.  The main challenge that these databases solve is how to handle massive amount of data at a reasonable cost and without poor performanc - distributed databases emerged to address this challenge and today we're seeing high adoption rate and quite impressive success stories such as the Netflix use of Cassandra/DataStax solution.  All that indicate the speed in which this market evolves.

 

 The need for a Big Data Application Platform

Application platforms provide a framework for making the development of applications simpler. They do this by carving out the generic parts of applications such as security, scalability, and reliability (which are attributes of a 'good' application) from the parts of the applications that are specific to our business domain.

Most of the existing application platforms such as Java EE and Ruby on Rails were designed to work with centralized relational databases in mind. Clearly, that model doesn’t fit well to the Big Data world simply because it wasn’t designed to deal with massive amount of data in first place. In addition to that, frameworks such as Hadoop are considered too complex as noted in VP/Research Director for Forrester Research Mike Gilpin, in his post, "Big Data" technology: getting hotter, but still too hard:

"Big Data" also matters to application developers - at least, to those who are building applications in domains where "Big Data" is relevant. These include smart grid, marketing automation, clinical care, fraud detection and avoidance, criminal justice systems, cyber-security, and intelligence.

One "big question" about "Big Data": What’s the right development model? Virtually everyone who comments on this issue points out that today’s models, such as those used with Hadoop, are too complex for most developers. It requires a special class of developers to understand how to break their problem down into the components necessary for treatment by a distributed architecture like Hadoop. For this model to take off, we need simpler models that are more accessible to a wider range of developers - while retaining all the power of these special platforms.

Other existing models for handling Big Data such as Data Warehouse don’t cut it either, as noted in Dan Woods' post on Forbes, Big Data Requires a Big, New Architecture:

...to take maximum advantage of big data, IT is going to have to press the re-start button on its architecture for acquiring and understanding information. IT will need to construct a new way of capturing, organizing and analyzing data, because big data stands no chance of being useful if people attempt to process it using the traditional mechanisms of business intelligence, such as a data warehouses and traditional data-analysis techniques.

To effectively write Big Data applications, we need an Application Platform that would put together the different patterns and tools that are used by pioneers in that space such as Google, Yahoo, and Facebook in one framework and make them simple enough so that any organization could make use of them without the need to go through huge investment.

Here's my personal view on how that platform could look like based on my experience covering the NoSQL space for a while now and through my experience with GigaSpaces.

Characteristics of Big Data Application Platform

As with any Application Platform, a Big Data application platform needs to support all the functionality expected from any application platform such as scalability, availability, security, etc.

Big Data Application platforms are unique in the sense that they need to be able handle massive amounts of data and therefore need to come with built-in support for things like Map/Reduce, Integration with external NoSQL databases, parallel processing, and data distribution services and on top of that, they should make the use of those new patterns simple from a development perspective.

Below is a more concrete list of the specific characteristics and features that define what Big Data Application Platform ought to be. I've tried to point to the specific Java EE equivalent API and how it would need be extended to support Big Data application.

Support Batch and Real Time analytics

Most of the existing application platforms were designed for handling of transactional web applications and have little support for business analytics applications.  Hadoop has become the de facto standard for handling Batch processing; Real Time analytics, however, is done through other means outside of the hadoop framework, mostly through an event processing framework, as I already outlined in details during my previous post Real Time Analytics for Big Data: An Alternative Approach.

A Big Data Application Platform would need to make Big Data application development closer to mainstream development  by providing a built-in stack that includes integration with Big Data databases from the NoSQL world, and Map/Reduce frameworks such as Hadoop and distributed processing, etc.

It also needs to extend the existing Transaction processing and Event Processing semantics that come with JavaEE for handling of Real Time analytics that fits into the Big Data world as outlined in the references below:

Making Big Data Application Closer to Mainstream development practices

• Domain model and Data access API

Writing Big Data application is quite different than writing a typical CRUD application to a centralized relational database in many forms. The main difference is with the design of our data domain model, and the API and Query semantics that well use to access and process that data.

Mapping has proven to be a fairly effective approach to map the impedance mismatch between different data models and API’s.  A good reference on that regard is the use of O/R mapping tools such as Hibernate for bridging similar impedance mismatch between Object and Relational models.

Abstraction frameworks such as Spring has also proven to be quite useful means to plug-in different data sources that doesn’t comply to common API through plug-in approach.

To make the development of Big Data Application closer to mainstream development Big Data application platform need to come with similar mapping and abstraction tools that would make the transition from the existing API’s significantly smoothers.

Today we already have various mapping and abstraction framework available.

For Big Data Mapping tools:

For batch processing were already seeing the increase adoption of frameworks such as Hive which provide an SQL like façade for handling complex batch processing with hadoop.

JPA provides more standard JEE abstraction that fits into Real Time Big Data application.    Google App Engine uses Data Nucleus on top of Big Table, With GigaSpaces we came up with a high performance JPA abstraction for our In Memory Data Grid using Open JPA, and Redhat came up with Hibernate OGM (Object-Grid Mapping). 

In addition to these, we're seeing that the existing mapping tools add extensions to support data partitioning semantics through annotations as in the case of EclipseLink and others.

For Big Data Spring Abstraction:

SpringSource came out with an interesting and even more high level abstraction known as Spring Data that makes it possible to map different data stores of all kinds into one common abstraction through annotation and plug-in approach. The abstraction is leaky currently, but shows progress in the space.

• Business logic

In Java EE, business logic refers to the parts in our application that are responsible for processing the data. As with JavaEE the data often lives in a centralized database SessionBean was often mapped to a single instance per user session.

With BigData the common pattern for processing data is through Map/Reduce. Map/Reduce was designed to handle the processing of massive amount of data through moving the processing logic to the data and distributing the logic in parallel to all nodes.  Having said that, developing parallel processing code is considered fairly complex.

A Big Data Application Platform would need to make Map/Reduce and parallel execution simple. One way of doing that is by mapping this semantics into existing programming models. One example would be to extend the current SessionBean model to support this sort of semantics (as with the GigaSpaces Remoting semantics) – this makes parallel processing look like a standard remote method invocation. Another way is to provide more native parallel execution semantics by extending existing semantics such as createNativeQuery in JPA, or the executor API in Java. 

Interestingly enough, at the time of writing this post I came across  Robin van Breukelen's post, "Distributed Fork Join," that shows how you can use the latest Java 7 fork/join API in similar context.

• Support new semantics that fits the dynamic web era

Clearly, SQL is a great query language (or else it wouldn't have taken over for database access!) but SQL also has its limits. When we're writing web content, we often deal with dynamic data structures that continuously evolve. When the amount of this data gets massive, as with Big Data, it becomes almost impossible to manage those changes through a standard SQL schema evolution model.

To address this need, BigData platforms need to add support schemaless semantics as a first class citizen. That often means that the data mapping layer mentioned earlier would need to be extended to support document semantics. An example for that is MongoDB, CouchBase, Cassandra and the new GigaSpaces Document API (which, as it turns out, maps to MongoDB and Cassandra very easily.)

• Provide built-in semantics for handling of the tradeoffs between consistency, availability, scalability rather than trying to force a least common denominator as with XA and JTA

JEE application platform forced a fairly strict consistency model through JTA, Distributed Transactions, and so forth. Unlike Java EE, Big Data Application platforms need to support more relaxed versions of those semantics that would enable flexibility between consistency, scalability, and performance. A good example on how that configuration may look like is covered in Cassandra/DataStax documentation and one of my previous post on the subject of CAP theory.

• Provide built in support for In-Memory data store to gain best performance and latency

RAM-based devices provide the best performance/latency results. Big Data platforms need to provide seamless integration between RAM and Disk based devices where data that is written in RAM would be synched into Disk asynchronously. In addition to that, they need to provide common abstractions that enable users to use the same Data access API for  both devices and thus make it easier to choose to the right tool for the job without changing the application code.

Good starting points on that regard are the JPA abstractions now being supported by In-Memory Data-Grids and NoSQL data stores as well as the Spring Data abstraction as noted above.

• Provide built-in support for event driven data distribution using pub/sub model

The current Java EE frameworks provide limited support for event processing through message-driven beans and JMS. For Big Data applications, we need to add data awareness to the current MDB model that make it easy to route messages based on data affinity and content of the message. We also need to provide more fine grained semantics for triggering events based on data operation (delete, update,..) as well as content as with CEP (Complex Event Processing). 

In addition to that we need to provide higher level abstractions that use the underlying pub/sub support to enable data synchronization and locality. A good example for that is the use of LocalCache and LocalView where LocalCache is useful for random access patterns by caching the data that was used most recently and LocalView which provides means to Cache a specific segment of data that is known in advance. Both LocalCache and LocalView use the underlying pub/sub support to ensure continuous synchronization of the data from the BigData pool by tracking changes on the Big Data pool and updating the local copy of the data implicitly.

Built in support for public/private cloud

Big Data applications tend to consume lots of compute and storage resources.  There are a growing number of cases where the use of the cloud enables significantly better economics for running  Big Data applications. To take advantage of those economics, Big Data Application Platforms need to come with built-in support for public/private clouds that will include seamless transition between the various cloud platforms through integration with frameworks such as JClouds. Cloud Bursting provides a hybrid model for using cloud resources as spare capacity to handle load. To effectively handle Cloud Bursting with Big Data well have to make the data available for both the public and private side of the cloud under reasonable latency – which often requires other services such as data replication.

Open & Consistent management and orchestration across the stack

A typical Big Data application stack includes multiple layers such as the database itself, the web tier,  the processing tier, caching layer, the data synchronization and distribution layer, reporting tools, etc. One of the biggest challenge is that each of those layers come with different management, provisioning, monitoring and troubleshooting tools. Big Data applications tend to be fairly complex by their very nature; the lack of consistent management,  monitoring and orchestration across the stack makes the maintenance and management of this sort of application significantly harder.

In most of the existing Java EE management layers, the management application assumed control of the entire stack. With Big Data applications, that assumption doesn’t apply. The stack can vary quite significantly between different application layers therefore the management layer of Big Data Application Platform needs to come with a more open management that could host different databases, web-containers etc., and provide consistent management and monitoring through this entire stack.

Final words

Java EE application servers played an important role in making the development of database-centric web application closer to mainstream. Other frameworks such as Spring and Ruby on Rails later emerged to increase the development productivity of those applications. Big Data Application Platforms have a similar purpose – they should provide the framework for making the development, maintenance and management of Big Data Applications simpler. In a way, you could think of Big Data Application platforms as a natural evolution of the current application platforms.

With the current shift of Java EE application platforms toward PaaS, we're going to see even stronger demand for running Big Data applications on cloud based environments due to the inherent economic and operational benefits. Compared to the current PaaS model, moving data to the cloud is more complex and would require more advance support for data replication across sites, cloud bursting etc.

The good news is that Big Data Application platforms are being implemented with these goals in mind, and you can already see migration yielding the benefits one should expect.

References

• NoSQL Job Trends – August 2011 (Dzone)
• Big Data" technology: getting hotter, but still too hard (Forrester)
• Big Data Requires a Big, New Architecture (Forbes)
• Real Time analytics for Big Data: Facebook's New Realtime Analytics Syste
• Example for Big Data application using Cloudify

Real Time analytics for Big Data: Facebook's New Realtime Analytics System

Recently, I was reading Todd Hoff's write-up on FaceBook real time analytics system. As usual, Todd did an excellent job in summarizing this video from Engineering Manager at Facebook Alex Himel, Engineering Manager at Facebook.

In this first post, I’d like to summarize the case study, and consider some things that weren't mentioned in the summaries. This will lead to an architecture for building your own Realtime Time Analytics for Big-Data that might be easier to implement, using Facebook's experience as a starting point and guide as well as the experience gathered through a recent work with few of GigaSpaces customers. The second post provide a summary of that new approach as well as a pattern and a demo for building your own Real Time Analytics system..

The Business Drive for real time analytics: Time is money

The main drive for many of the innovations around realtime analytics has to do with competitiveness and cost, just as with most other advances.

For example, during the past few years financial organizations realized that inter-day risk analysis of their customers' portfolios translated to increased profit as they could react faster to profit and loss events.

The same applies to many of the online ecommerce and social sites. Knowing what your users are doing on your site in real time and matching what they do with more targeted information transforms into better conversion rate and better user satisfaction, which means more money in the end.

Todd provides similar reasoning to describe the motivation behind Facebook's new system:

Content producers can see what people like, which will enable content producers to generate more of what people like, which raises the content quality of the web, which gives users a better Facebook experience.

Why now?

Real time analytics goes mainstream

The massive transition to online and social applications makes it possible to track user patterns like never before. The correlation between the quality of data that providers track and their business success is closely related: for example, e-commerce customers want to know what their friends think about products or services, right in the middle of their shopping experience. If sites cannot keep up with their thousands of users in real-time, they can lose their customers to sites that can.

So while risk analytics in financial industry was still a fairly small niche of the analytics market the demand for real time analytics in Social, eCommrce and SaaS applications brought the demand for real time analytics to main-stream business under massive load.

No one has time for batch processing anymore.

Technology advancement

Newer infrastructures and technologies like tera-scale memory, nosql, parallel processing platforms, and cloud computing, provide new ways to process massive amount of data in shorter time and with lower cost.  As most of the current analytics systems weren’t built to take advantage of these new technologies and capabilities, they haven't been able to adopt to real time requirements without massive changes.

Hadoop Map/Reduce doesn’t fit the real time business

One of the hottest trends in the analytics space is the use of Hadoop as almost the de-facto standard for many of the batch processing analytics application. While Hadoop (and Map/Reduce in general) does a pretty good job in processing massive logs and data through parallel batch processing, it wasn’t designed to serve the real time part of the business.

Strong evidence for that can be seen from the moves of those who were known as the “poster children” for Map/Reduce: Google and Yahoo both moved away from Map/Reduce. Google has moved to  Google Percolator for its indexing service. Yahoo came-up with a new service S4 which was designed specifically for real time processing.

It is therefore not surprising that Facebook reached the same conclusion as it relate to Hadoop:

Hadoop/Hive..Not realtime. Many dependencies. Lots of points of failure. Complicated system. Not dependable enough to hit realtime goals.

 Facebook's Real Time Analytics system

According Todd,  Facebook evaluated a fairly long list of alternatives including as MySQL, an in-memory cache, Hadoop Hive/Map Reduce. I highly recommend reading the full details from Todd's post.

I tried to outline Facebook architecture based on Todd's summary in the diagram below:

Every user activity triggers an asynchronous event, through AJAX – this event is logged in a tail log using Scribe.  Ptail is used to aggregate the different individual logs into a consolidated stream. The stream is batched in 1.5 sec groupings by Puma which stores the event batch into HBase. The real time logs are kept for a certain period of time and than get cleared from the system.

Obviously this description is a fairly simplistic view – the full details are provided in the original post.

Evaluating the Facebook Architecture

Facebook reasoning behind their technology evaluation seems acceptable for the most part, although there are some obvious concerns.

There were two things that caught my attention:

Memory Counters

(Facebook) felt in-memory counters, for reasons not explained, weren't as accurate as other approaches. Even a 1% failure rate would be unacceptable. Analytics drive money so the counters have to be highly accurate.

It sounds to me that the evaluation was based on memcached. By default, it's not highly available; a failure would result in loss of data. Obviously, that doesn’t apply to some other memory based solutions such as In Memory Data Grids (for example, GigaSpaces and Coherence both were designed for high resiliency.)

Cassandra vs HBase

The choice of HBase over Cassandra was also very interesting mainly since Cassandra was developed by another Facebook team to address write scalability.  The choice had to do with the write rate differences between the two alternatives at the time of evaluation:

HBase seemed a better solution based on availability and the write rate. Write rate was the huge bottleneck being solved.

Eric Hauser  posted a comment on this analysis which seem to indicate that this issue has been addressed with Cassandra:

When Facebook engineers started the project 6 months ago, Cassandra did not have distributed counters which is now committed in trunk. Twitter is making a large investment on Facebook for real-time analytics (see Rainbird). Write rate should be less of of bottleneck for Cassandra now that counter writes are spread out across multiple hosts. For HBase, every counter is still bound by the performance of single region server? A performance comparison of the two would be interesting

Eric's comment is indicative of how dynamic the NoSQL space is. I’d be interested in how different the technology selection would be now.

The Architecture

There are a few common principles that drive the architecture for this type of system:

• Events logging needs to be extremely fast, to minimize the latency impact on the site.
• Events need to be reliable, otherwise the entire system's accuracy is questionable and the data is devalued.
• The real time data in the form of logs is kept for a certain period of time (x hours or x days)
• Write scalability is key.
• Post processing can happens in batches, the size of the batch depends on how *real-time* we need this data to be.
• Writes to a backend database need to be done asynchronously.

Facebook seem to follow those same principles in their architecture while keeping the system at scale at a fairly impressive rate.

There are a few questions that are still open as I don’t have full visibility into their system:

• Are Ptail and Puma centralized components? If so, don't they pose a potential bottleneck? Based on Todd's summary and Alex' presentation, it seems that the way Facebook scaled their system is by splitting the PTail stream by categories of events so each type can be handled by a different data center.
• Puma batches event logs in memory before it writes them into Hbase – what happens if Puma fails before the batch is written?
• The solution seemed to be limited to handling simple counters. This seem to be a fairly severe limitation, as many systems need to produce more complex relationships even during the real time part of the system, as also indicated as part of the future enhancements, as shown:

"(We need to know) how many people across a time window liked a URL. Easy to do in MapReduce, hard to do with a naive counter solution.”

• In one point, it is noted that Facebook chose not to rely on memory for counters. However, throughout the description it seems that there is still strong reliance on keeping data within memory boundaries:

“(We) write extremely lean log lines. The more compact the log lines the more can be stored in memory..”

“(We) batch for 1.5 seconds on average. Would like to batch longer but they have so many URLs that they run out of memory when creating a hashtable”

Looking backward – wouldn’t it be better to store the data in-memory in the first place? Why add the extra architecture components, if you're able to make memory work for you? This is a crucial question, of course, because it focuses attention on memory availability. As mentioned, though, there are in-memory data grids that are designed for just this kind of situation.

• It is noted that Puma writes it batches to Hbase sequentially:

“(We) wait for last flush to complete for starting new batch to avoid lock contention issues...”

What if this rate is lower than the actual write rate?

Realtime.Analytics.next

Interestingly enough, I was asked to draw a solution for similar challenge for a voice recording system: collect lots data from various sources and process them in real time. The good news is that it's doable, as shown by Facebook's success.

The better news is that it's actually fairly easy. We can add rich query capabilities, elastic scaling, database and platform neutrality, evolution of data, and more without making things unnecessarily difficult - it's not so much that this is the next generation of realtime analytics as much as map/reduce and the hbase approach used by facebook is the previous generation.

I'll describe how it can be done more simply in the next post.

References

• Video: Realtime Analytics for Big Data: A Facebook Case Study
• Real Time Analytics for Big Data: An Alternative Approach
• Big data in real time is no fantasy (GigaOm)
• Facebook's New Realtime Analytics System: HBase To Process 20 Billion Events Per Day
• Read/write scale without complete re-write
• How Map/Reduce and the Cloud are Affecting Analytics Applications
• Is MapReduce going mainstream?
• Memory is the New Disk for the Enterprise

Read/write scale without complete re-write

Last week I was attending one of our Partner events in Stockholm where I presented the convergence of trends in the data scalability world – specifically the transition from NoSQL to NewSQL and the convergence of trends that brings the existing SQL and new NoSQL world much closer together as I noted in a previous post, "YeSQL: An Overview of the Various Query Semantics in the Post Only-SQL World."

During the event, Ronnie Bodinger, Head of IT at Avanza Bank AB, gave an excellent talk on how they turned their existing online banking application into a new site that was designed for read/write scaling.

Avanza's System Description:

Avanza Bank is a  Swedish bank that makes it easy for investors to make equity transactions and fund switches. It runs the most trades on the Stockholm Stock Exchange.

It prides itself for providing advanced tools for its investors through its online banking system.

The current online system is a typical web site based on Java/JSP and Spring.

Scaling architecture of the existing site

Most of the interaction with the current site are read-mostly, the main scaling challenge was scaling on concurrent read operations. Read scaling was addressed through a side-cache architecture that is common with many of the existing LAMP + Memcached deployments where the first query hits the database and the following queries hit the cache.

The New System

The new site was designed to fit into the real-time and social era.  This means that a lot of the traffic and activities are now being generated by user activities and not by just by the site owner. This activities need to be presented in real-time to the bank users.

Challenges

The changes to the new site lead to a significant change in the traffic and load behavior that drives a new class of scaling challenges:

Write scaling

In case where there are lots of updates involved the existing side-cache architecture leads to diminishing returns as the cache becomes obsolete pretty quickly and therefore synchronizing the cache only adds overhead.

Using Oracle RAC with a high end hardware platform didn’t prove itself either and yielded a fairly expensive solution that didn’t meet the scaling requirements.

Unlike “Green Field” applications, Avanza has an existing online application (a “Brown Field”) that serves its current customers. That brings the following list of additional challenges :

• Existing Data Model

The entire data model of the application was designed for a relational model – changing the data model or moving it to a new NoSQL architecture as was considered would involve a huge change that could turn into a potentially years-long effort.

• Legacy system

The online bank application consists of large set of legacy application and third party services. Re-writing the existing infrastructure is either impossible (due to the dependency on third party tools) or impractical.

 

• Complex environment

As it often the case, a large portion of the legacy applications weren’t designed for scale, and weren’t built with a clear holistic architecture as they were built through layers over the years. This increases the complexity of scaling by quite a bit.

 

• Existing Skillset

The existing development team already had fairly good knowledge of Java and Java EE. Changing the team and/or developing a completely new skillset is a huge barrier as the ramp-up time required to bring new developers up to speed with the complexity of the system can take years.

The solution: Read/write scale without complete re-write

It was clear that meeting the new scaling challenges would involve changes to the existing application – the main question was how to scope that change so that it wouldn’t require a complete re-write. The second goal was to build the change in a way that would reduce the TCO of the current system.

To achieve those two goals the following approach was used:

• Minimize the change by clearly Identifying the scalability hotspots

The areas of the application that need intensive write access are often small parts of the overall system.  The first step would therefore be to minimize the change to only the hot spots of the application and keep the rest of the application un-touched. In the case of Avanza, the hot spots were identified on certain tables used by the online web application. Most of the backend systems were still accessing the database for reporting,  synchronization and batch processing and could therefore remain unchanged.

• Keep the database as is

One of the key piece in the current design is the ability to address the read/write scalability outside of the database context (See the next bullet). This makes it possible to keep the existing database and the schema of the data unchanged. In that way, all the rest of the systems continue to work with the database as if nothing changed.

• Put an In Memory Data Grid as a front end to the database

Scaling the application is done by front-ending the application with an In-Memory-Data-Grid (IMDG). The IMDG contains all the hot tables or rows of the original database. The online web application accesses the IMDG instead of the database. The IMDG is distributed in nature thus allowing scalability by distributing the load over a cluster of machines for both read and write operations.

• Use write-behind to reduce the synchronization overhead

Updates from the IMDG to the underlying database is done asynchronously in batches through an internal queuing mechanism (redo-log).

• Use O/R mapping to map the data back into its original format

In many cases to achieve best scalability, we need to partition the data. This often involves changes to the data schema. Changing the data schema could break the entire system including the areas that don’t suffer from the scalability bottleneck. To meet this impedance mismatch, we scope the data schema changes only to the IMDG. The data is mapped from the IMDG schema to the original schema through standard O/R mapping tools such as Hibernate or OpenJPA.

• Use standard Java API and framework to leverage existing skillset

One of the challenges with many of the new NoSQL database alternatives is that they often force a complete re-architecture.  This comes with the a fairly high cost of re-building the skillsets within the organization across the board for developing against new APIs as well as for maintaining capacity and sizing of those new databases.

IMDGs such as GigaSpaces expose standard APIs, such as JPA. In addition, they allow organizations to extend the use of the existing database while removing a large part of read/write load – both the use of standard APIs and existing database enables organizations to leverage their existing skillset and still meet their scalability requirements. It also enables them to take a smoother transition (through baby steps) into a completely new scale-out architecture by allowing a plug-in new scalable database at a later stage.

• Use two parallel (old/new) sites to enable gradual transition

Switching all the customers into a new system at once is often a bad idea. A better approach would be to enable gradual transition of selected customers into the new sites. A common model to achieve that would be to run two parallel sites. The challenge in doing so is the synchronization between the two parallel sites. In the case of Avanza, they used the GigaSpaces Mirror service to synch all the changes from one site to the other and in that way keep the two sites up to date.

The diagram below provides a visual summary of this approach:

The TCO angle

The second goal in the project was to reduce the TCO of the current system.

This is achieved in the following way:

• Use RAM for high performance access and disk for long term storage

As I noted in one recent post, a RAM based solution can be 10x – 100x cheaper than Disk based solution for high performance applications.

In addition to that, the price for RAM goes down continuously.

1GB can cost only 1.9$ a month..  , storing 1T bytes of data completely in RAM can fit into a single RAC with total cost of $1.8k per month.

The optimal solution would therefore be to use RAM to manage data that needs high speed write/read access and disk-based storage for the long-term data that is accessed less frequently.

• Use commodity Database and HW – A single instance of an Oracle RAC deployment could reach to $500k. Putting a data-grid in front of the database enables us to remove the needs for many of the high-end features available in the Oracle RAC database. It also enables us to remove the need for high end hardware such as storage devices, infiniband network etc.

In this specific case, it was possible to move the data into MySQL and use commodity Dell machines to run the entire relational data system.

 

Final words

Many existing applications are built with layers on top of layers on top of layers of development. Relational databases often sit at the heart of those systems. Scaling these systems is therefore an extremely challenging task. This leads a lot of organizations to take the easy route, simply paying more for more high-end hardware and databases. We have reached to the point where this approach doesn’t cut it for many cases - and it's simply too expensive to maintain in the face of cheaper, more scalable alternatives.

In a world where the impact of software accretion is no longer tolerable, it's clear that a chnage is inevitable if we're to meet the new demands for scalability. The real question is how to make that change.  The approach that was taken by Avanza Bank in their use of GigaSpaces is an excellent reference: it shows that they were not just able to meet the new scaling requirements fairly quickly through measurable and easy small changes, but they also reduced their cost of ownership significantly as noted recently by Ronnie Bodinger's statement:

“Throwing out Oracle's sluggish databases increases Avanza performance of their business-critical systems, while reducing license costs significantly."

..

"Today we have a large Oracle Cluster, which costs a lot. The new system goes in a fraction of it."

You can read the full article here (Note the article is in Swedish – use Google Translator to read the English version of the article..)

The recorded version of Ronnie presentation as it was captures through one of the audience camera is available here.  

Incidentally, I'd like to offer a sincere word of thanks to So4It, who did an excellent job in working with Avanza and organizing the partner event.

Scaling Social Ecommerce: Architecture Case study

In February this year I had the opportunity to meet Tomer Gabel, an Application Engineer at SHC Israel, who gave an excellent talk at one of our customer road shows about Social E-Commerce and the scalability challenges associated with handling of social graphs. Tomer later joined me for a case study, published on InfoQ.

In this post I tried to summarize the main takeaway from the interview. Enjoy..

What is Social E-Commerce anyway?

According to wikipedia, Social Shopping is a mechanism for e-commerce where shoppers' friends become involved in the shopping experience. Social shopping attempts use technology to mimic the social interactions found in physical malls and stores.

The opportunity

A recent study showed that over 92 percent of executives from leading retailers are focusing their marketing efforts on Facebook and subsequent applications. Furthermore, over 71 percent of users have confirmed they are more likely to make a purchase after “liking” a brand they find online. (source)

I recently came across an interesting analysis on the Subscribers, Fans, & Followers: The Social Break-Up. The report analyzed social behavior, and more specifically why consumers end brand relationships. Even though the focus of this post is not about ending brand relationships, I thought that some of the statistics presented in this report are quite useful for quantifying where we are today with social e-commerce in terms of the market size and its potential.

Facebook statistics:

• 73% of U.S. online consumers have created a profile on Facebook.
• 65% of U.S. online consumers are currently active on Facebook.
• 42% of U.S. online consumers (64% of those on Facebook) are “FANS” (have “liked” a company on Facebook).

Twitter statistics:

• 17% of U.S. online consumers have created a Twitter account.
• 9% of U.S. online consumers are currently active on Twitter.
• 5% of U.S. online consumers (56% of those on Twitter) are FOLLOWERS  (use Twitter and have “followed” at least one company).
• 71% of FOLLOWERS expect to receive marketing messages from companies through Twitter.

Clearly the adoption of social commence is quite staggering.

The focus of this post wasn’t to convince you should invest in social commerce but to give a context for one of the common scalability challenges associate with building social ecommerce platform - the "Social Graph."

The Social Graph challenge

The Social Graph is “the global mapping of everybody and how they're related.”

Processing social networks is not an easy proposition:

• Massive amounts of branching data
• No data locality
• Very few assumptions can be made about the data

In other words, to meet the capacity scaling demand you can’t store the data in a centralized location. That forces us to distribute the data. On the other hand, to access (or query) the data, we can’t assume that the data for our query is located in a single node. That forces us to look for the data on multiple nodes.

Let’s take a simple scenario to get some sense of the complexity of the problem:

 Example:

• Imagine every Facebook user (500 million)
• Imagine each person is only connected to 100 others (conservative estimate)

Query: How is user X connected with Y?

• X has 100 friends
• Each of them has 100 friends
• 10,001 nodes visited!
• 101 reads from the underlying storage system!

Clearly even if we can scale at the capacity level we can’t scale at the query level.

The crux of the problem:

• High branch factor necessitates many loads to serve even a simple request
• No data locality + high branch factor means very high random I/O
• Traditional storage models (RDBMS, flat files etc.) are a poor fit

The Solution:

• Use Memory as the main storage
  • Random I/O access works much better on memory devices than on disk (See more details here)
• Execute the code with the data - Using Real Time Map/Reduce
  • To reduce the number of iterations required to execute a particular query we use the executor API. The executor API enables us to push the code to the data. By doing that we can execute fairly complex data processing on the data node at memory speed vs network speed.
• De-normalize the data
  • To reduce the amount of traversal access and network hops per query on the graph we need to copy elements of the graph into each node. For example the list of Friends and friends of friends (up to a certain degree) could be stored in each node and thus become available to any element of the graph  without the need to consult with other nodes.

The operational perspective

Scaling Social Ecommerce involves not just the application architecture but also the operational aspect. This is of particular importance as ecommerce and social sites tend to evolve quite rapidly and therefore the time it takes to release a new feature from development to production is critical.  This process is often referred to as Continuous Deployment or in our specific case it can be referred to also as  "Continuous Scaling."

There are basically two factors that are important to achieve continuous scaling/deployment:

• Automation – if the process of deployment and scaling involves lots of human intervention than the time it would take to release a new feature would get significantly higher. It is therefore important that the entire process can be fully automated. In some cases, you may want to have manual check points in the process but even in that case the expectation is that manual intervention would involve clicking on a <continue> button and nothing else.  To achieve this level of automation we need to interact with the our application infrastructure through an API. This process of automation and integration between the development and operational environment is often referred to as DevOps. The GigaSpaces reference for dev-ops API is provided here.
• Schema evolution –There are many cases in which we would want to add or change our existing data structure to our existing application as part of the upgrade process without brining the system down. This is considered one of the more complex challenges to deal with with many of the existing data bases. A document model is schema-less and therefore more suited for this sort of rapid data change.

 The full story

Tomer did a a great job describing in more details the scalability challenges and the approach that they have taken in Sears to address those challenges as well as providing the specific insight on the performance figures that they achieved with their current system. The full interview recording is provided below:

 

References

• Designing a Scalable Twitter
• Memory is the New Disk for the Enterprise
• Why Existing Databases (RAC) are So Breakable!
• The Common Principles Behind the NOSQL Alternatives