Nati Shalom's Blog

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

A interesting note on Google Megastore & CAP,..

I was reading James Hamilton's coverage on Google Megastore: The Data Engine Behind GAE (a must read!), where he summarizes the main architecture assumption behind Google Megastore.

The thing that caught my eyes was the following line:

Support for consistency unusual for a NoSQL database but driven by (what I believe to be) the correct belief that inconsistent updates make many applications difficult to write (see I Love Eventual Consistency but …)

This was later further elaborated upon in one of James' comments below the posting:

The cap theorem is undeniably true but, like many engineering issues, it requires judgment. Consistency can be maintained over remarkably a large scope of data. The availability that CAP forces to be given up certainly exists but can be made quite small. Overall, I would say that CAP encourage many to give up hope way too quickly and build an eventually consistent system when better choices are available with more work. The paper under discussion here shows that a high-scale store can be fully consistent over substantial volumes of data and they wisely decided to not make the entire store globally consistent. Interpreting and applying CAP is a judgement call and we see one approach here. SimpleDB chose to do both (http://perspectives.mvdirona.com/2010/02/24/ILoveEventualConsistencyBut.aspx) which is along the lines you recommend in your conclusion: “as scalable data sources mature they will increasingly and perhaps ubiquitously allow user-selectable CAP choices in much the same way that SQL stores almost always allow user-selectable transaction consistency”. I think you are right.

Interestingly enough James also points to another interesting part of his previous post on the subject (I love eventual consistency but...) where he describes how Amazon deals with consistency tradeoffs with SimpleDB:

What I’ve learned over the years is that strong consistency, if done well, can scale to very high levels. The trick is implementing it well. The naïve approach to achieve full consistency is to route all updates through a single master server but clearly this won’t scale. Instead divide the update space into a large number of partitions, each with its own master. That scales but there is still a tension between the number of partitions and the cost of maintaining many partitions and avoiding hot spots. The obvious way to avoid hot spots is to use a large number of partitions but this increases partition management overhead. The right answer is to be able to dynamically repartition to maintain a sufficient number of partitions and to be able to adapt to load increases on any single server by further spreading the update load.

I found those comments particularly interesting as many people refer to Google and Amazon as the poster children for justifying eventual consistency.

It is interesting to see that the reality is that even Google and Amazon - which I would consider the extreme cases for big data - realized the limitation behind eventual consistency and came up with models that can deal with scaling without forcing a compromise on consistency as I also noted in one of my recent NoCAP series here, here and here.

Speaking of Score Card..

They say that timing is everything in life.. this time was no exception.

Yesterday I published my score card for 2nd generation PaaS and how it differs from Application Servers.

A few hours later, I received an email with a presentation from one of our new customers that is going to present at our GigaSpaces Roadshow.

This slide caught my attention:

 

* Note that GS stands for GigaSpaces.

The customer is a large ISV (1900 employee, $3B in revenue...). Their new product is a real-time BI system that needs to process events in real time from multiple data sources -– social media included (most other alternatives do that in batch mode) and produce actionable reports on customer activities. They had fairly challenging scaling, latency, and cost requirements:

• Massive Scale, 100M Customers, a billion events a day
• Business real time < 100ms
• Low cost 0.05 customer/year

Now it is not often that a company the size of GigaSpaces competes with giants like IBM and Oracle on one of their core IPs and leaves them so far behind.

How was this possible?

Well, our advantage is in our disadvantage –- size. Let me explain.

Removing the Silos

GigaSpaces as a case study

As a small company, we have to be extremely efficient at what we do.  We adopted Lean methodology and Scrum across the board, and we stick to that model quite religiously.  Every feature goes through millions of automated tests. We manage to build GA-quality releases every two weeks.

On the product side, we have designed our product in a way that follows the principles that I laid out in my previous post, i.e. -- without Silos. In GigaSpaces everything is mapped to data. Tables, Queues, even method invocations are mapped to the same underlying data structure representation. 

We maintain ONE cluster throughout the entire stack to manage this scalability and high availability of the data.  ALL of our APIs are simple facades on top of our shared data cluster (a.k.a space). This means that when we optimize performance, fix bugs, improve scalability, and add new features -- it applies immediately through the entire stack, including the .Net and C++ parts of our product.

The fact that all of our APIs share the same data opened new opportunities for optimization that weren’t possible before. For example, we can avoid a lot of the synchronization overhead that is associated with passing a message and then updating it. We can also mesh the driftnet APIs  without unnecessary copies.

As an example, I can choose to use the key/value store API for low latency transaction processing and the SQL API for analytics over the same data that was produced through the key/value APIs.

Because of our size, we couldn’t afford the high maintenance and support that often comes with the type of deployments that we deal with. As a result, we were forced to develop lots of built-in automation and troubleshooting tools as an integrated part of our product, and later we integrated it with the development framework and application in such a way that every application is made operationally ready from its inception.

Now, the nice thing in all this is that both we and our customers benefit from that efficiency. We gain more customers without increasing our costs. Our customers get a better quality product, short fix cycles, and more features faster.

The latest 8.0 release is the best evidence of this, IMO. We managed to deliver 3 times more features than what we normally accomplished in previous releases, and all that with only a minor increase in our development team.

 *A Little Promotion*

If you're in London on the 2nd of February , Paris on 3rd, San Francisco on 7th, or NYC on the 9th, I’d like to welcome you to join our Roadshow and hear more first-hand stories from these customers (Dow Jones, Sears, Amdocs, and many more top-tier organizations). See you there!

You can register and read the full details here.

NoCAP - Part II Availability and Partition tolerance

Abstract

Eric Brewer's CAP theorem is one of the foundations behind the design and architecture of many of the large scale systems architectures. According to the one form of the CAP theorem, “You either choose availability or consistency. You cannot choose both.” 

Is this our only choice?

In this post I'll try to provide an aggregated view of the different perspectives on this question and plant the seeds for an alternative approach.

Availability vs Partition tolerance.

One of the thing that comes up repetitively through the various debates is the lack of clarity behind the definition between Availability and Partition Tolerance.

To set the stage I’ll  start by referring to the original definitions by Gilbert and Lynch . I’ll use Coda Hale's You Can't Sacrifice Partition Tolerance - a "must-read" that provides a good summary.

Brief history  (By Coda Hale):

In 2000, Dr. Eric Brewer gave a keynote at the Proceedings of the Annual ACM Symposium on Principles of Distributed Computing¹ in which he laid out his famous CAP Theorem: a shared-data system can have at most two of the three following properties:Consistency,Availability, and tolerance to network Partitions. In 2002, Gilbert and Lynch²converted “Brewer’s conjecture” into a more formal definition with an informal proof.

The original definition by Gilbert and Lynch:

Availability

For a distributed system to be continuously available, every request received by a non-failing node in the system must result in a response. That is, any algorithm used by the service must eventually terminate … [When] qualified by the need for partition tolerance, this can be seen as a strong definition of availability: even when severe network failures occur, every request must terminate.

Partition tolerance

In order to model partition tolerance, the network will be allowed to lose arbitrarily many messages sent from one node to another. When a network is partitioned, all messages sent from nodes in one component of the partition to nodes in another component are lost. (And any pattern of message loss can be modeled as a temporary partition separating the communicating nodes at the exact instant the message is lost.)

Dr. Stonebraker’s "Consistency over Partition Tolerance"

Dr. Michael Stonebraker’s post Errors in Database Systems, Eventual Consistency, and the CAP Theorem argues that since partition failures are rare you might sacrifice partition tolerance for consistency and availability:

Obviously, one should write software that can deal with load spikes without failing; for example, by shedding load or operating in a degraded mode. Also, good monitoring software will help identify such problems early, since the real solution is to add more capacity. Lastly, self-reconfiguring software that can absorb additional resources quickly is obviously a good idea.

In summary, one should not throw out the [Consistency aspect] so quickly, since there are real error scenarios where CAP does not apply and it seems like a bad tradeoff in many of the other situations.

Henry Robinson (Cloudera) describes "CAP Confusion"

Henry Robinson, in his response to Dr. Stonebraker entitled Problems with ‘partition tolerance’, argues that failure are inevitable and therefore you can’t give away partition tolerance:

..Partition tolerance is not something we have a choice about designing into our systems. If you have a partition in your network, you lose either consistency (because you allow updates to both sides of the partition) or you lose availability (because you detect the error and shutdown the system until the error condition is resolved). Partition tolerance means simply developing a coping strategy by choosing which of the other system properties to drop. This is the real lesson of the CAP theorem – if you have a network that may drop messages, then you cannot have both availability and consistency, you must choose one..

Coda Hale : "You Can't Sacrifice Partition Tolerance"

Coda Hale also argues that you can’t give away partition tolerance in his response, You Can't Sacrifice Partition Tolerance. Mr. Hale suggest a more relaxed version of availability through graceful degradation that is based on the yield and harvest model from  Eric A. Brewer's ACM article, Lessons from Giant-Scale Services:

You cannot ... choose both consistency and availability in a distributed system.

Of the CAP theorem’s Consistency, Availability, and Partition Tolerance, Partition Tolerance is mandatory in distributed systems. You cannot not choose it. Instead of CAP, you should think about your availability in terms of yield (percent of requests answered successfully) and harvest (percent of required data actually included in the responses) and which of these two your system will sacrifice when failures happen.

Dr. Stonebraker - Partition tolerance and Machine failure are two separate things

Both Henry Robinson and Coda Hale reffed to a Machine failure as part of their argument that you can't avoid partition tolerance . Dr. Stonebraker provides clarification on that point on his response to Coda Hale: 

.., the dead node is in one partition and the remaining N-1 nodes are in the other one. The guidance from the CAP theorem is that you must choose either A or C, when a network partition is present. As is obvious in the real world, it is possible to achieve both C and A in this failure mode. You simply failover to a replica in a transactionally consistent way. Notably, at least Tandem and Vertica have been doing exactly this for years. Therefore, considering a node failure as a partition results in an obviously inappropriate CAP theorem conclusion.

Daniel Abadi - Problems with CAP

Daniel Abadi Assistant Professor of Computer Science at Yale University outlines four issues in the current definition of CAP in Problems with CAP, and Yahoo’s little known NoSQL system:

• ..The definition of CP looks a little strange --- “consistent and tolerant of network partitions, but not available” --- the way that this is written makes it look like such as system is never available --- a clearly useless system

• ..the roles of the A and C in CAP are asymmetric. Systems that sacrifice consistency (AP systems) tend to do so all the time, not just when there is a network partition

• ..What if there is a network partition? What does “not tolerant” mean? In practice, it means that they lose availability if there is a partition. Hence CP and CA are essentially identical

• Lack of latency considerations in CAP significantly reduces its utility.

Daniel suggest an alternative definition to CAP:

..CAP should really be PACELC --- if there is a partition (P) how does the system tradeoff between availability and consistency (A and C); else (E) when the system is running as normal in the absence of partitions, how does the system tradeoff between latency (L) and consistency (C)?

My Take

The main issue that I see with the current definition of CAP is that it describes each of the properties of CAP in absolute terms. In reality, however, each of the properties can be applied in various degrees. Eric Brewer's definition of yield and harvest is one example of a more relaxed version of availability. Interestingly enough the last note in Gilbert and Lynch is also an interesting recognition that it may be possible to achieve different tradeoffs that provides both Availability and Consistency:

..in partially synchronous models it is possible to achieve a 
practical compromise between consistency and availability. In particular, 
most real-world systems today are forced to settle with returning “most of 
the data, most of the time.” Formalizing this idea and studying algorithms 
for achieving it is an interesting subject for future theoretical research.

The same applies to Partition Tolerance – there are various degrees of partition tolerance. Probably the most extreme one are the one suggested by some of the commenters which is to do with WAN. As Dr. Stonebraker argues, even that is not necessarily a high likelihood scenario for most applications:

There is enough redundancy engineered into today’s WANs that a partition is quite rare. My experience is that local failures and application errors are way more likely. Moreover, the most likely WAN failure is to separate a small portion of the network from the majority. In this case, the majority can continue with straightforward algorithms, and only the small portion must block. Hence, it seems unwise to give up consistency all the time in exchange for availability of a small subset of the nodes in a fairly rare scenario.

So the question whether or not you can address partition tolerance shouldn’t be measured in absolute terms but against the most likely scenario for your application. The answer as to what is a most likely partition tolerance scenario really comes down to common sense and may vary over a course of time and application business needs. 

The introduction of the cloud may also change some of our core assumptions on how we deal with failure. With cloud, we can easily bring an alternate machine up to deal with a failure in matters of minutes, and we can also assume that we have a fairly robust underlying infrastructure with lots of built-in redundancy. That in itself changes the scenario where failures can happen as well as the way to deal with them, compared to the time when the original CAP Theorem was written. 

Another source for confusion is the use of Amazon, Google and Facebook as a reference to justify the eventual consistency model. What we often tend to forget is that Amazon, Facebook and Google face fairly unique challenges that are not that common and even those three still rely on strongly consistent systems for the majority of their applications. As noted in Will Scalable Data Stores Make NoSQL a Non-Starter?:

Facebook famously invented the NoSQL Cassandra database but still relies on the venerable MySQL-plus-memcached combination for the brunt of its critical operations.

The bottom line of my argument is that giving up consistency should be our last resort. Rather then giving up Consistency for Partition Tolerance we could consider a different set of tradeoffs that deals with various degrees of Consistency, Availability, and Partition tolerance that fit our business needs and also take the performance and latency tradeoffs into account.

In the next post on that subject I’ll outline how that can be achieved using the reference architecture that I outlined in the previous post.

Special thanks

I would like to thank Ron Pressler for the constructive debate and for the useful references 

References:

• Errors in Database Systems, Eventual Consistency, and the CAP Theorem (Dr. Michael Stonebraker)

• You Can't Sacrifice Partition Tolerance - (Coda Hale)

•  Will Scalable Data Stores Make NoSQL a Non-Starter?

• Problems with CAP, and Yahoo’s little known NoSQL system

• You Can't Sacrifice Partition Tolerance

 

Marrying memcached and NoSQL

Abstract:

Memcached is one of the most common In-Memory cache implementation.  It was originally developed by Danga Interactive for LiveJournal, but is now used by many other sites as a side cache to speed up read mostly operations. It gained popularity in the non-Java world, too, especially since it’s a language-neutral side cache for which few alternatives existed.  

As a side-cache, Memcache clients relies on the database as the system of record, The database is still used for write,update and complex query operations.  Since the  memcached specification includes no query operations, memcached is not a database alternative, unlike most of the NoSQL offerings. It also exclude memcache from being a real solution for write scalability. As a result of that many of the heavy sites started to move away from Memcache and replace it with other NoSQL alternatives as noted in a recent highscalability post MySQL And Memcached: End Of An Era?

The transition away from memcached to NoSQL could represent a large investment as many sites are already heavily invested in memcached usage. In this post, I'll illustrate an alternative approach in which we’ll extend the use of memcache for write scaling, add other goodies such as high availability and elasticity by plugging GigaSpaces as the backend datastore, and avoid the need for a re-write. The pure Java implementation could also be seen as a benefit as it can increase the adoption of memcached within the Java community and leverage the portability of java to other platforms,

Memcached overview

The diagram below shows a typical use of Memcache. It outlines a simple deployment topology often referred to as a side cache. This topology is very popular to address read mostly scenarios.

Typical use of Memcache for scaling read mostly applications

In this mode, read operations first check the memcached servers for an artificial key, derived from the requested data. If the key doesn’t exist on the memcached server, a query is made to the database, and the result is then stored with that artificial key into the memcached instances. Subsequent calls with that query go through the memcached deamon and thus saves access to the database. Updates remove the data from the memcached instances, then write the data to the database and the memcached instance at the same time in the most common situation. Wikipedia has a code snippet that provides a more detailed example of this in action.

The main benefit of this model lies in its simplicity. The memcached API is extremely simple to use for this specific scenario. The fact that it relies on a simple and open protocol as opposed to a rich client bound to to a specific language and implementation makes it portable across a wide variety of languages and environments. It also known to be fairly scalable for handling read operations as due to the inherited share nothing model.

The advantages of memcached that make it so simple and popular are the exact same things that make it fairly limited for any scenario that goes beyond read-mostly/side-cache scenarios. Some of those limitations were the main driver that forced some popular sites to switch to a NoSQL as an alternative to memcached as noted by Digg and Twitter.

From Looking to the future with Cassandra:

The fundamental problem is endemic to the relational database mindset, which places the burden of computation on reads rather than writes. This is completely wrong for large-scale web applications, where response time is critical.

From Cassandra @ Twitter: An Interview with Ryan King:

We have a lot of data, the growth factor in that data is huge and the rate of growth is accelerating. We have a system in place based on shared mysql + memcache but its quickly becoming prohibitively costly (in terms of manpower) to operate. We need a system that can grow in a more automated fashion and be highly available.

Summarizing the points from Digg and Twitter, the main limitations of the current Memcached implementations are the lack of support for:

• Write scalability
• Elasticity
• High availability

I would also add to that list the lack of a consistency model, the limited query support, a client-based optional sharding model, management, and monitoring.

Marrying memcached and NoSQL

Many of the limitations of memcached outlined above are addressed by various NoSQL/In-Memory-Data-Grid implementations. The main difference is that many of the NoSQL alternatives were designed to act as a database and not just as a side-cache. As such they were geared for write-scaling and come with built-in elasticity and high availability in mind.

The main motivation behind the integration between the two is to address the current limitations of memcached without forcing a complete and expensive re-write. Another motivation is that we can leverage the rich client and language support that comes along with the memcached protocol as a simple way to extend the use of NoSQL alternatives to other languages.  

While the pattern of integrating Memcache with existing NoSQL backend is fairly generic, in this post i would refer specifically to the GigaSpaces integration.

GigaSpaces memcached support

Joseph Ottinger published a good summary of the rationale behind the GigaSpaces memcache support in a post on TheServerSide.com, “Did Someone Say GigaSpaces Now Has Memcached Support?” In this post, I'll try to provide a reference to the underlying GigaSpaces implementation in the context of the memcached usage pattern that I outlined above.

The GigaSpaces memcached support is fairly simple. It consists of a listener service (written in Java) that implements the memcached protocol for client applications, and maps the protocol into equivalent GigaSpaces operations. There are three mode of operations in which the integration works:

Memcached-compatible – reliability and write-scaling

The memcached-compatible mode is set to be compatible with the exact same way that memcache client works today.

In this mode, each GigaSpaces node runs an embedded instance of the memcached service. The communication between the memcache client and the GigaSpaces node is handled through the memcached protocol.  The GigaSpaces memcached service communicates with the GigaSpaces backend directly in-memory.

By using GigaSpaces as the backend datastore we gain immediate benefits from the robustness and richness of the GigaSpaces environment. This includes write-scaling and reliability through the built-in database integration for pre-loading data from a database and write-behind for storing data back into the database. In addition to that we gain built-in management, monitoring and deployment automation (through the GigaSpaces dev-ops API).  The fact that the entire stack is built in Java also enables to leverage the Java tool ecosystem (debugging tools, monitoring tools, profilers) as well as the portability of the Java Virtual Machine.

As the memcached protocol doesn’t include dynamic discovery, the memcached clients have static binding to the backend server, which doesn’t provide GigaSpaces’ elasticity and dynamic scaling.

Memcached load-balancer – elasticity + write-scaling, reliability

In this mode of operation, the memcached server instance does not use the embedded GigaSpaces instance but instead would reference a remote GigaSpaces cluster. In this mode, the GigaSpaces memcached server instance becomes a memcached load-balancer between the memcached clients and the GigaSpaces data partitions. Each GigaSpaces cluster could have many memcached-load-balancers. As with HTTP load-balancers, the clients only need to point to a single IP address, and every memcached request is mapped to a GigaSpaces partition according to the key used in any given operation.

The main benefit of this mode is its simplicity and elasticity. Elasticity is achieved by the fact the GigaSpaces cluster can grow as needed. The GigaSpaces proxy will discover those instances dynamically and route the memcached requests to the appropriate node. The same applies to fail-over i.e. if one of the GigaSpacs nodes fails the proxy will rote the request to a hot backup automatically. 

The memcached load balancer can be also hosted as a service within the GigaSpaces container. To gain even better scalability of the load-balancer we can combine the memcached-compatible mode (our first example)  with this load-balancer mode where each of node in the gigaspaces cluster could have the embeded memcached service configured as a load-balancer. In that configuration, a client can succeed by selecting one of the nodes in the cluster while gaining access to the entire cluster.

The downside of this approach is performance as each memcache request will to go through two network hops. For that purpose we often set GigaSpaces LocalCache to minimize that extra hop for read operations.  In this way subsequent get operation on the same key would be resolved within the load-balancer itself. Any update on that key through any of the partition will be updated automatically on the local cache as well.

Another potential use of that mode is gateway between two network domains.  It is much easier to cross firewall and network domains through a single gateway then having the entire cluster available.

Memcached RPC – Map/Reduce support, multi-lang data feeder

One of the interesting and lesser-known uses of the memcached protocol is for Remote Procedure Calls or Map/Reduce operations. In this mode, a set operation on the memcache server will be translated to a command, and a get operation is translated into a return value for that command. This is extremely useful for data-feeder scenario and aggregated query scenarios. Non-Java feeders can also leverage the GigaSpaces support for dynamic languages to pass in code that will get executed in the server close to the data.

The execution of the code would be done through the use of the GigaSpaces polling container. In this case we would set the polling container to poll for the memcache entries.

NOTE: We may be releasing a new version with built-in support for this specific pattern so the API is still subject to change ..

Simple Example:

The following example illustrates the first two modes outlined in this post – the memcached-compatible and the load-balancer modes.

For the sake of simplicity we will use a single memcached server instance. A more advanced deployment of a complete cluster is referenced at the end of this post.

Running a single memcache instance:

In this example we will run a GigaSpaces instance using the gs-memcached{.sh/.bat} . The gs-memcached utility takes one argument that points to Space URL.

> gs-memcached <Space URL>

Note for gs-memcached utility is available only since  GigaSpaces 8.0 release. For earlier versions use puInstance utility instead using the following format:

> puInstance(.sh/.bat}" -properties space embed://url=”<Space URL>” "<gigaspaces root>\deploy\templates\memcached"

Running a single memcache in compatible mode:

To run a memcached server that reference to an embedded GigaSpaces instance we will use a Space URL in the following form: “/./<name>”

In our example we will use “/./memcache” which basically set a memcache server that points to an embedded GigaSpaces instance under the name “memcached”.

>gs-memcached{.sh/.bat} ”/./MyMemcached”

or simply

>gs-memcached{.sh/.bat}

which puts a default name “/./memcached” as the Space-URL

Running a single memcached load-balancer:

To run a memcache in a load-balancer mode we would reference to remote GigaSpaces cluster by setting the Space-URL to a remote cluster as outlined below:

> gs-memcached{.sh/.bat} ”jini://*/*/remoteGigaSpaces”

A simple memcache client:

There is nothing really unique in our memcached client. We can use any standard memcached library and point it to the host:port of our server. The default port is set to11211.

   1: MemcachedClient memClient=new XMemcachedClient("localhost",11211);

   2: memClient.set("1", 3600*60, "some value");

   3: memClient.get("1");

   4: memClient.delete("1");    

Monitoring and Management:

One of the benefits of using GigaSpaces as the backend data store is the fact that we can leverage the existing monitoring and management to monitor the statistics and activity of our memcache deamon. We can use any of the GigaSpaces management tools for that purpose, including the rich client UI, the web-based UI, and command line and API based management and deployment tools.

The diagram below shows the real time statistics gathered from the gs-ui utility.

Advanced setup:

In a more advanced setup, we would want to leverage the GigaSpaces Dev-Ops API and SLA driven container to automate the deployment and managing the SLA of that deployment. A detailed description of that mode is provided in the GigaSpaces memcached documentation.

Final words – Qcon next week

We often tend to look at any new technology as like a sort of new religion. Quite often, our needs change and with that we tend to switch from one technology (religion) to another. Memcached and NoSQL are a good example for such a transition. I personally believe that to gain the benefit of a new technology we don’t need to completely abandon previous technologies but instead learn from them to make the new technologies and techniques better. That is particularly true when it comes to data technology - consider SQL, which been around for more then 4 decades!

A new writeup that came out last week on GigaOM, “Will Scalable Data Stores Make NoSQL a Non-Starter?,” is just a reminder of how quickly technology tends to shift. That makes the importance of taking a more evolutionary path toward the revolution that we’re going through even more important than ever before.

During the Qcon session Yes, SQL!  next week Uri Cohen will lay out some of the various NoSQL query languages that are available today starting with memcached, MongoDB, Cassandra, Redis and outline a model  for leveraging the best out of each in our existing JEE world. I hope it will lead to an interactive learning session and debate, as it often does at QCon. 

References

• Memcached (Wikipedia)
• MySQL And Memcached: End Of An Era? (highscalability.com)
• Did Someone Say GigaSpaces Now Has Memcached Support?  (TheServerSide.com)
• GigaSpaces memcache documentation. (GigaSpaces.com)
• GigaSpaces FREE Community edition (GigaSpaces.com)
• YeSQL: An Overview of the Various Query Semantics in the Post Only-SQL World
• GigaSpaces dev-ops API (GigaSpaces.com)
• Will Scalable Data Stores Make NoSQL a Non-Starter? (GigaOM)
• Cassandra @ Twitter: An Interview with Ryan King (twitter.com)
• Looking to the future with Cassandra (Digg.com)

NoCAP

 

In the past few months i was involved in  many of the NoSQL discussions. I must admit that i really enjoyed those discussions as it felt that we finally started to break away from the “one size fit it all” dogma and look at the data management solutions in a more pragmatic manner. That in itself sparks lots of interesting and innovative ideas that can revolutionize the entire database market such as the introduction of document model, map-reduce and new query semantics that comes with it. As with any new movement we seem to be going through the classic hype cycle. Right now it seems to me that were getting close to the peak of that hype. One of the challenges that i see when a technology reaches its peak of the hype is that people stop questioning the reason for doing things and jump on new technology just because X did that.  NoSQL is no different on that regard.

 

 

In this post i wanted to spend sometime on the CAP theorem and clarify some of the confusion that i often see when people associate CAP with scalability without fully understanding the implications that comes with it and the alternative approaches.

 

I chose to name this post NoCAP specifically to illustrate the idea that you can achieve scalability without compromising on consistency at least not at the degree that many of the disk based NoSQL implementations imposes.

 

Recap on CAP

 

Quoting the definition on wikipedia:

The CAP theorem, also known as Brewer's theorem, states that it is impossible for a distributed computer system to simultaneously provide all three of the following guarantees:^[1][2]

• Consistency (all nodes see the same data at the same time)
• Availability (node failures do not prevent survivors from continuing to operate)
• Partition Tolerance (the system continues to operate despite arbitrary message loss)

CAP and NoSQL

Many of the disk based NoSQL implementations was originated from the need to deal with write scalability. This was largely due to the changes in traffic behavior that was mainly a result of the social networking  in which most of the content is generated by the users and not by the site owner.

In a traditional database approach achieving data consistency requires synchronous write to disk and distributed transactions (known as the ACID properties).

It was clear that  the demand for write scalability would conflict with the traditional approaches for achieving consistency (synchronous write to a central disk and distributed transactions).

The solution to that was:  1) Breaking the centralized disk access through partitioning of the data into distributed nodes. 2) Achieve high availability through redundancy (replication of the data into multiple nodes) 3) Use asynchronous replication to reduce the write latency.

The assumptions behind point 3 above is going to be the center in this specific post.

 

Graphical representation of the Cap Theorem. (Source)

 

The Consistency Challenge

One of the common assumptions behind many of the NoSQL implementations is that to achieve write scalability we need to push as many operations on the write-path to a background process in order that we could minimize the time in which a user transaction is blocked on write.

The implication is that with asynchronous write we loose consistency between write and read operations i.e. read operation can return older version then that of write.

There are different algorithms that were developed to address this type of inconsistency challenges, often referred to as Eventual Consistency.

For those interested in more information on that regard i would recommend looking at Jeremiah Peschka post Consistency models in nonrelational dbs. Jeremiah provides a good (and short!) summary of the CAP theorem, Eventual Consistency model and other common principles that comes with it such as (BASE – Basically Available Soft-state Eventually, NRW, Vector clock,..).

Do we really need Eventual Consistency to achieve write scalability?

Before I'll dive into this topic i wanted to start with quick introduction to the term “Scalability” which is often used interchangeably with throughput. Quoting Steve Haines:

The terms “performance” and “scalability” are commonly used interchangeably, but the two are distinct: performance measures the speed with which a single request can be executed, while scalability measures the ability of a request to maintain its performance under increasing load

(See previous post on that regard:  The true meaning of linear scalability)

In our specific case that means that write scalability can be delivered primarily through point 1 and 2 above ( 1-Break the centralized disk access through partitioning of the data into distributed nodes. 2-Achieve high availability through redundancy and replication of the data into multiple nodes) where point 3 ( Use asynchronous replication to those replica’s to avoid the replication overhead on write) is mostly related with write throughput and latency and  not scalability. Which bring me to the point behind this post:

Eventual consistency have little or no direct impact on write scalability .

To be more specific my argument is that it is quite often enough to break our data model into partitions (a.k.a shards) and break out from the centralized disk model to achieve write scalability. In many cases we may find that we can achieve sufficient throughput and latency just by doing that.

We should consider the use of asynchronous write algorithms to optimize the write performance and latency but due to the inherited complexity that comes with it we should consider that only after we tried simpler alternative such as using db-shards, FLASH disk or memory based devices.

Achieving write throughput without compromising consistency or scalability

The diagram below illustrates one of the examples by which we could achieve write scalability and throughput without compromising on consistency.

As with the previous examples we break our data into partitions to handle our write scaling between nodes. To achieve high throughput we use in-memory storage instead of disk. As in-memory device tend to be significantly faster and concurrent then disk and since network speed is no longer a bottleneck we can achieve high throughput and low latency even when we use synchronous write to the replica.

The only place in which we’ll use asynchronous write is the write to the long-term-storage (disk).  As the user transaction doesn’t access the long-term storage directly through the read or write path, they are not exposed to the potential inconsistency between the memory storage and the long-term storage. The long-term storage can be any of the disk based alternatives starting from a standard SQL databases ending with any of the existing disk based NoSQL engines.

The other benefit behind this approach is that it is significantly simpler. Simpler not just in terms of development but simpler to maintain compared with the Eventual Consistency alternatives. In case of distributed system simplicity often correlate with reliability and deterministic behavior.

 

Final words

It is important to note that in this post i was referring mostly to the C in CAP and not CAP in its broad definition.  My points was not to say don’t use solution that are based on CAP/EventualConsistency  model but rather to say don’t jump on Eventual Consistency based solutions before you considered the implications and alternative approaches. There are potentially simpler approaches to deal with write scalability such as using database shards, or In-memory-data-grids.

As were reaching the age of Terra-Scale devices such as Cisco UCS where we can achieve huge capacity of memory, network and compute power in a single box the area’s in which we can consider to put our entire data in-memory get significantly broader as we can easily store Terra bytes of data in just few boxes. The case of Foursquare's MongoDB Outage is interesting on that regard.  10gen's CEO Dwight Merrimanargued that the entire set of data actually needs to be served completely in-memory:

For various reasons the entire DB is accessed frequently so the working set is basically its entire size Because of this, the memory requirements for this database were the total size of data in the database. If the database size exceeded the RAM on the machine, the machine would thrash, generating more I/O requests than the four disks could service.

It is a common misconception to think that putting part of the data in LRU based cache ontop of a disk based storage could yeild better performance as noted in the sanford research The Case for RAM Cloud

..even a 1% miss ratio for a DRAM cache costs a factor of 10x in performance. A caching approach makes the deceptive suggestion that “a few cache misses are OK” and lures programmers into con-figurations where system performance is poor..

In that case using pure In-Memory-Data-Grid as a front end and disk based storage as long term storage could potentially work better and with significantly lower maintenance overhead and higher determinism. The capacity of data in this specific case ( <100GB)  shouldn't be hard to fit into single UCS box or few of the EC2 boxes.

 

References

• Lessons from Pat Helland: Life Beyond Distributed Transactions

• The true meaning of linear scalability

• Consistency models in nonrelational dbs

• Foursquare's MongoDB Outage

• CAP theorem

• Why Existing Databases (RAC) are So Breakable!

• Problems with CAP, and Yahoo’s little known NoSQL system

• CAP Confusion: Problems with ‘partition tolerance’

• You Can't Sacrifice Partition Tolerance
• Gilbert and Lynch. Brewer’s conjecture and the feasibility of consistent, available, partition-tolerant web services. (2002)
• DeCandia et al. Dynamo: Amazon’s highly available key-value store. (2007)