GigaSpaces

July 13, 2009

Platform as a Service: The Next Generation Application Server?

Platform as a Service is a term that can be fairly confusing for many people. Normally the term is associated with Google App Engine from Google and Force.com from Salesforce.com as the main references for this model. From a technical point of view, it is aimed to provide a similar type of value to the one that is currently provided by many of the application servers i.e. it provides a generic container that can host different applications and shield them from the details of the underlying operating system, network, database implementation. Unlike most of the existing application servers it was designed for massive scaling from day one. Another big difference is in the way it is being consumed. With PaaS you don’t need to install any software and go through all the hoops to setup a cluster environment etc. PaaS is provided as a hosted service that is pre-configured and installed. You get a production ready environment right at the start.

Key Characteristics of a Cloud/SaaS Enabled Application Platform

Last week I came across David Mitchel Smith presentation from Gartner. David provided a good definition to the main PaaS characteristics. A snippet from his presentation covering this area is given below:


image

What’s interesting is the great emphasize on Multitenancy support. The fact that the platform is going to be shared between multiple applications and potentially even different customers require various levels of tenancy support and isolation to ensure that even though were taking advantage of the fact that we can share resources between applications, each application needs to be able to use the platform as if it is running on its own dedicated resources. Another interesting point is the need for XTP support. Many would view XTP as a niche that is normally referenced in the high end part of the market, so it is fair to ask why would XTP fit into a general PaaS solution?  XTP represents a model for supporting enterprise transaction processing applications in an extremely scalable environment. In our case, scalability is not necessarily driven from the demand of a particular application but from the fact that many applications are going to run on a shared environment.  A PaaS targeted to enterprise applications would need to provide support for this level of scalable transaction processing support as a core service.

Can you run your existing business applications with GAE or Force.com?

No. Unfortunately as in anything in life reality can really spoil the party.

Force.com was designed to make it simple to run business applications that are database centric i.e. CRM, Reporting etc. It provides a rich set of high level services that make building such applications extremely simple. Google provides a more generic application platform and recently announced support for Java which is a big step towards reducing vendor lock-in concerns.  Google seems to be geared for consumer based applications. Force.com offers a more high level platform that is based on its proprietary services. This means that in order to take advantage of their service you will need to go through a complete re-write.

Force.com is based on a database centric architecture. They also seem to be limited in scalability as they partition their database per application. This means that if your application needs to scale more than what a single database can provide, you can find yourself pretty much locked.

Google's recent support for Java makes their offering closer to standard JEE application servers, however their current support impose a lot of limitations due to their sandbox model. These limitations mean that at the end of the day GAE can be applicable only to a small set of relatively simple applications.  Since there is no guarantee or control over the resources that you are going to receive from their underlying infrastructure, it is likely that the application performance will be unpredictable and will therefore be affected by other applications that are sharing the same hardware.

The fact that the platform as a service shields you from the details of the underlying infrastructure is what makes it simple and that is both the advantage and limitation. You can’t control the environment, you can’t choose your operating system, you can’t install your own set of services and you can’t control the performance characteristics of this platform.

This puts a huge adoption barrier for most current enterprises.

IaaS vs PaaS

Infrastructure as a Service provider provides a hosted service model that offers plain machine level access. This model is also known as Server as a Service. The fact that you get access to the bare metal allows you to run almost any application on this hosted environment. Unlike PaaS, IaaS gives you extreme flexibility. You can choose your own operating system, install any package that you want, you can setup your firewall, security etc. Amazon is known to be the leader in this space and also provides  a set of services on top of their infrastructure such as SimpleDB, SQS and MapReduce. Having said all that, this flexibility comes at the cost of complexity. In many cases you will need to install your own software, configure it, tune it etc. before you could make it run effectively on the cloud. Many application developers don’t have the skill-set to do that. This exposes many operational challenges as many organizations are not geared to support this type of environment from their own IT. The high level services provided by Amazon are still proprietary and would require a complete re-write if you plan to use such services.

PaaS for Enterprise applications – doing it right

The ideal solution would be to combine the best of the two worlds i.e. the flexibility of IaaS and simplicity of PaaS, and here is how:

  • Build a Generic PaaS on top of AWS – To build a PaaS we don’t need to re-invent the wheel. Unlike Google and Force.com we don’t need to own the infrastructure, we can actually use Amazon infrastructure or even better build our PaaS such that it can be portable between Amazon IaaS and a VMware IaaS. By doing so, the PaaS can provide us the ability to deploy applications in a simple way just as in GAE but would still enable us to control the environment, install our own software and get the full flexibility that IaaS provides.
  • JEE as first class citizen – Many of the existing enterprise applications are built in JEE. Making JEE a first class citizen within our PaaS environment will enable you to leverage the existing skill-sets within those organizations. Similar to the standard model that has more than one implementation out there, reducing the lock-in factor significantly.
  • Pre-configured for extreme scalability – All the services provided through the PaaS will need to be implemented and pre-configured for extreme scaling and come with a production ready setup to enable dynamic scaling and fault-tolerance.

Next generation application Server?

Yes, In my opinion, PaaS represents the next wave in middleware technology. One that is targeted for virtualized enterprises, one that was designed for scale-out from the get go, one that fits the new way of delivering SaaS applications and one that can be extremely simple to use. The current PaaS players i.e Google and Salesforce and to a lesser degree Microsoft represent one type of player, those that own it all, i.e. the infrastructure and the platform. There is already a new emerging category of players in the PaaS market, Application PaaS players. Application PaaS players would specialize on delivering only the platform and not the hardware and network infrastructure. They act as the bridge that will enable portability between different infrastructure providers including the internal IT (a.k.a private-cloud). The application PaaS players will also be segmented in similar ways to the way application servers are segmented today, i.e there would be the one targeting the low-end consumer market in similar ways to Google App Engine, the ones that will be aimed toward the high end of the market and those who will be specialized in certain languages or development framework, i.e. Ruby/ Java/ .Net etc. In the PaaS type of world, those who would be able to provide a holistic solution that works smoothly across all the application tiers would have an advantage over those who are providing point solutions.

What about my existing applications ?

Most people would categories PaaS as a platform that is delivered through the internet. That statement would make the idea behind PaaS irrelevant for a large part of the existing enterprise applications, as a majority of them are not ready to run their application in a hosted service over the internet. Let’s examine that statement:

Many of the existing IT run farms of application servers in their internal IT. Those application servers are running in an internal data center that is not that different from any other hosted services, only that it is a specialized hosted service tailored for the needs of the specific organization. I would therefore argue that those organizations that are already running such application server farms would find it easier to evolve such server farms to PaaS model than to change their entire IT infrastructure into internal cloud. The reason is that those applications were already written to run in an application container model, therefore a large part of the transition work can be done within the container implementation and outside the application code. Targeting them first would therefore be potentially easier transition than trying to transform all your other applications into a virtualized environment.

I know that there are many people out there that would argue that internal PaaS is not a PaaS because it doesn’t answer the exact definition of PaaS. However in my view the similarities exceed the differences and the value to the enterprise would be almost identical to the one that I would receive by any of the internet based PaaS platform.

Will I lose control?

PaaS will provide the internal IT much better control over the applications that are running on their environment. You can have high visibility as to ways the application consume resources. In the same way, you can control fairly tightly the application security, scalability, resource management and fault-tolerance which can finally be managed in a consistent way across all the applications.

Final words

Platform as a Service represent the next generation application server IMO. GAE and Force.com are most known references in the market for that model. Both tries to offer the complete stack and that’s both their advantage and limitation. There already is a new category of Application PaaS providers that specialize on providing PaaS on top of existing infrastructure providers. There would be generic PaaS providers (similar to GAE) geared for the low end part of the market and those that are geared for the high end of the market. There would also be more vertical PaaS providers (Similar to Force.com)  i.e. those that will provide PaaS for a certain segment of applications or segment of the market such as Online Gaming PaaS, Telco PaaS etc. A good example for such a service is Twilio. As I outlined in my recent post (Google App Engine plus Amazon AWS: Best of both worlds), this is not just a theory but a reality in the making. GigaSpaces' contribution to this new reality is our new cloud framework on Amazon EC2. Geva Perry provides an excellent overview of other good examples that follows that same line his post What's Really Exciting About Cloud Computing.

References:

July 09, 2009

No to SQL? Anti-database movement gains steam – My Take

Eric Lai published a provoking article on Computerworld magazine titled “No to SQL? Anti-database movement gains steam” where he pointed to many references in which different Internet-based companies chose an alternative approach to the traditional  SQL database. The write-up was driven from the the inaugural get-together of the burgeoning NoSQL community who seem to represent a growing Anti-SQL database movement.

Quoting Jon Travis from this article:

Relational databases give you too much. They force you to twist your object data to fit a RDBMS [relational database management system],

The article points to specific examples that led different companies such as Google, Amazon, Facebook to choose an alternative approach. I outlined below what i found to be the main drivers behind that trend:

  • Demand for extremely large scale:

“BigTable, is used by local search engine Zvents Inc. to write 1 billion cells of data per day.”

  • Complexity and cost of setting up database clusters:

“PC clusters can be easily and cheaply expanded without the complexity and cost of ‘sharding,’ which involves cutting up databases into multiple tables to run on large clusters or grids.”

  • Compromising reliability for better performance:

“There are different scenarios where applications would be willing to compromise reliability for better performance. A good example for such a scenario is HTTP Session data. In such a scenario the data needs to be shared between various web servers but since the data is transient in nature (it goes away when the user logs off) there is no need to store it in persistent storage.”

Having said all that, it seems that many still agree that despite all the limitations of traditional database solution, SQL database are probably not going away:

“It's true that [NoSQL] aren't relevant right now to mainstream enterprises," Oskarsson said, "but that might change one to two years down the line.”

The current "one size fit it all" databases thinking was and is wrong

The article seem to point to an interesting trend where a growing number of application scenarios cannot be addressed with a traditional database approach. This realization is actually not that new. In  2007 I wrote a summary of Michael Stonebrake’s article, "One size fits all: A concept whose time has come and gone" on my blog: Putting the database where it belongs. The great thing is that It looks like this “old” news is spreading to the larger community. This can be explained by the continuous growth of data volumes, together with the growing need to process larger amounts of data in a shorter time. These two trends force many users to think of an alternative approach to the traditional database. The classic early adopters are those who hit the wall. It is very likely that as these alternative solutions mature, they will find their way into mainstream development as well.

Not your mom and dad’s database

The article seems to over glorify some of the alternatives that where mentioned while downplaying their limitations. A good example is Amazon SimpleDB. I wrote in the past a post about this, Amazon SimpleDB is not a Database, where I outlined some of the limitations of the Amazon SimpleDB solution. As you can see from these limitations, SimpleDB cannot and shouldn’t be positioned as a direct alternative to your existing database.

While I share many of the thoughts and enthusiasm of the anti-SQL movement, I would highly recommend taking very cautious steps toward any of the alternative solutions. It is very important that you make yourself familiar with their strengths and weaknesses, and avoid hitting their limitations at a point in time when you have very little room to maneuver. I’ve seen several cases where users developed their data model in a centralized model and expected that it will scale seamlessly once they switch to a partitioned topology. The fact that you can switch between centralized and partitioned topologies without changing your code doesn’t mean that your application will behave correctly and will scale as you expect.

This topic has actually been the center of a discussion in Architect Summit meeting we had last summer, which was hosted by eBay:

Abstractions and Partition Awareness
A horizontally-partitioned system typically provides an abstraction that makes the partitions appear as a single logical unit. eBay and Flickr, for example, both use a proxy layer to route requests by a key to the appropriate partition, and applications are unaware of the details of the partition scheme. There was near-universal agreement, however, that this abstraction cannot insulate the application from the reality that there partitioning and distribution is involved. The spectrum of failures within a network is entirely different from failures within a single machine. The application needs to be made aware of latency, distributed failures, etc., so that it has enough information to make the correct context-specific decision about what to do. The fact that the system is distributed leaks through the abstraction.

My recommendation would be that you design your data model to fit into a partitioned environment even if during the initial stage you’re still going to use a single centralized server. This will allow you to scale when you need to, without going to through a massive change.

What about in-memory alternatives?

An option that i found missing from this article and becomes fairly popular with many large websites is the use of an in-memory data store, In-Memory-Data-Grids as they are often called, such as Memcached, GigaSpaces, Coherence, eXtremeScale etc. With this model we front-end the database with an in-memory cluster which becomes the system of record and uses the SQL database as the background persistent store. For those looking to build social network graphs, real time events (as in Twitter), real time analytics, fraud detection, session management, etc., that is probably the more natural choice. Todd Hoff from highscalability.com wrote a very good article on this subject: Are Cloud Based Memory Architectures the Next Big Thing? I also wrote a detailed description how this approach works with GigaSpaces and MySQL: Scaling Out MySQL.

What about ACID transactions, consistency etc?

The traditional 2PC (two phase commit) model in which consistency is achieved through a central transaction coordination server is not going to fit with many of the distributed data management alternatives. In an earlier post, “Lessons from Pat Helland: Life Beyond Distributed Transactions,” Pat Helland suggested an alternative model to distribute transactions, a workflow model. Instead of executing a long transaction and blocking until everything is committed, break the operation into small individual steps where each step can fail or succeed individually. By breaking the transaction into small steps, it is easier to ensure that each step can be resolved within a single partition, thus avoiding all the network overhead associated with coordinating a transaction across multiple partitions. This has been one of the core concepts in designing scalable applications with Space Based Architecture (SBA). The Actor model that was introduced with new functional languages like Scala and Erlang is built into the SBA model, with the difference that in SBA, actors can share state and pass events by references, and thus avoid the overhead of copying the data with every transaction.

Shay Banon wrote a good description on how the Actor model works with SBA in this post:

image

“The above is a diagram of a simple simple polling container that wraps a service (Actor, the Order service in our example). The polling container takes (removes) events (Data), process them, and writes back Data to the collocated Space (Data Grid) it is running with.”


The need for real time data processing – a real life example

In the past weeks we had been involved with a prospect who is looking to add a social network to his eCommerce site. One of the requirement for this new service was the need to build a graph that includes friends of friends, and products in catalogs. The process for building that graph with an In-Memory-Data-Grid took 2-3 msec vs. tens of seconds with a traditional approach (note that part of the complexity in this case is that the query itself is ad-hoc and can’t be easily partitioned). Building that graph on-the-fly at these speeds just couldn't be done with traditional SQL database.

The reason that enabled us to get to this level of performance was:

  1. We kept the data in-memory.
  2. A large part of the complex query was pushed to where the data is (you can think of it as a modern alternative to stored procedure).
  3. We used partitioning to spread the data and leverage the accumulated memory capacity of those memory instances.
  4. We used both a scale-out and scale-up model to parallelize the query against all instances and take full advantage of multi-core as well as multi-machine power.
  5. We reduced the number of network hops by pushing the the heavy data manipulation to where the data is and by returning only the accumulated result over the network.

Final words

In summary I would say:

  • SQL databases are not going away anytime soon.
  • The current "one size fit it all" databases thinking was and is wrong.
  • There is definitely a place for a more a more specialized data management solutions alongside traditional SQL databases.

The adoption of these new solutions would be very much determined by two main factors:

  1. How well they integrate with the “existing SQL world.”
  2. How easy it will be to develop for these new alternatives, and how smooth a transition a given solution can offer for the average developer.

This is an area of continuous innovation that has been keeping us fairly busy in the past few years, and will probably continue to keep us busy. I’ll leave the details on how we deal with these two challenges to a separate post.

References:

June 25, 2009

Google App Engine plus Amazon AWS: Best of both worlds

George Lawton wrote a a good summary of my JavaOne talk in his article titled Google App Engine plus Amazon AWS: Best of both worlds 

Google App Engine (GAE) is focused on making development easy, but limits your options. Amazon Web Services is focused on making development flexible, but complicates the development process. Real enterprise applications require both of these paradigms to achieve success… What we really want is the flexibility and performance of AWS and the simplicity and ease of use of GAE.

This is exactly what we had been working on for the past year, leading us to the launch of our new cloud platform. With this platform we leverage GigaSpaces XAP as the high performance scale-out application server and Amazon as the robust and flexible IaaS. Together they form an alternative Platform as a Service geared for enterprise grade applications. This allows the cloud environment to inherit the extreme performance, latency and scalability of the XAP platform, which in turn enables achieving your performance and scaling target with less machines, implying a lower cost.


Real-life case study: Primatics financial – Risk analysis as a service

Francis de la Cruz and Argyn Kuketayev from Primatics Financial joined me through the presentation. In their part of the session they described their experience in developing a SaaS application for Real Time analytics.

Kuketayev described how Primatics used this approach to create a new automatically scaling cloud version of an existing banking application. Primatics initially developed a mortgage securities application that allows banks to estimate the value of a basket of hundreds of thousands of loans. The value of these loans fluctuates as economic conditions change and some portion of home owners cannot afford to make payments on their loans. Banks normally only need to assess the value of these loans at the end of each month, making them an ideal candidate for cloud services like AWS.

From a scalability perspective the challenge is to be able to provide a highly multi-tenant application that need to serve many firms, many users in that same firm each running many jobs at the same time. Implementing such a model can be fairly complex as you will need to be able to manage the life cycle of each job and each user independently and in isolation from one another.


The need for scale 

Trying to build such a service directly on Amazon is going to be fairly complex, as you can learn from George’s summary below:

Primatics wrote the first version of EVOLV:Risk as a hosted web application for a regional bank.. The application needed to be fault tolerant so that if one node crashed, they did not have to restart the application over again from the beginning. Kuketayev said that it is not just about the loss of four hours, but the office is trying to close out the month and needs to access data to end the monthly cycle so they can go home.

Using GigaSpaces' toolset they rewrote the entire application infrastructure in about four-months to run on top of AWS. Now they can kick off as many instances as required for different banking customers, and each instance runs significantly faster than before. Kuketayev said that it is important for banks that none of their applications run on the same infrastructure as another bank.


The diagram below shows the specific architecture that Primatics ended up using. Those that are familiar with Space Based Architecture would find it fairly straight forward:

The application is built out of  a set of processing units. Each processing unit contains the compute agents in the form of a polling-container.  The compute agents gets a a reference to a remote Data Grid that is shared by all processing units. Each agent gets the job injected to it by the polling container and gets a reference to the data it required to process the job. Once the job is completed, the result is stored back in the space. The results are flashed out back to a database through a mirror service.

In a case of a failure, other compute agents are able to continue from the exact point of failure and continue the job processing as if nothing happened. This is because the state of the job is kept safe in the data-grid and not in the agent’s memory.


image

Kuketayev from Primatics nicely summarized thye lesson he learned after going through the experience of trying to build it on his own vs. trying to use GigaSpaces:

Kuketayev said that one of the biggest lessons is that you need to have your infrastructure do the provisioning for you automatically, or otherwise you end up spending a lot of time just turning things on and off. He said they are now using configuration APIs to automate this process, whereas before they were using scripts. This allow for automatically throttling and failover recovery without human intervention.

Kuketayev advised "You need to make sure you use the right tools … You don't want to have to worry about provisioning and reliability. Make sure you have provisioning, failover, monitoring and SLA out of the box."


The full JavaOne presentation is available here:


Final words

Fr solution providers the size of Primatics, building a risk analysis application as a service couldn’t be possible without cloud computing. Cloud enabled them to offer their solution as a service without the need to go through major investment of building a data center to support it.

Primatics’ experience is not special. One of the benefits of building Software as a Service is that you have one shared environment for all your customers. At the same time, one of the challenges is that in a shared environment, failure becomes more public and will impact ALL your clients. If the system doesn’t scale well, you’re going to be hit twice as hard as in a standalone application.

Building a robust and scalable SaaS application can be fairly complex. A good cloud infrastructure will get you a first class data center, but it won’t solve your application requirements.What’s interesting with cloud computing is that it forces you to think about the cost and efficiency of your application more than ever before. In the Primatics example, running a simulation of 100 nodes for 3 hours is very likely to fail at some point. A failure during such a simulation will immediately cost you 300 hours, not to mention the fact that you might lose the simulation window for the day and the reputation challenge you’ll will be facing with your customers. In addition, putting the data in-memory and making the application run 3-5 times faster means that you would need 1/5 of the machine power, which saves 80% of the cost of running the application.

I believe that the challenges imposed by cloud computing force us to focus on what we do best and avoid investing in areas which are not core to our business. Because the pay-per-use model significantly lowers the cost barrier, going down the path of writing your own infrastructure, as many have tried to do before, will be much more expensive and risky then ever before.

References:

June 01, 2009

GigaSpaces Launches a New Version of its Cloud Computing Framework

We have been working for a while on our new Cloud Computing Framework (CCF). We wanted to wait until we had some real customers and partners behind the platform in production before spreading the word on a larger scale. And now, I am very happy to report that we have reached this point sooner than expected and just in time for the JavaOne event.

The new version includes an enhanced integration with the Amazon billing system Devpay designed to enable users to use GigaSpaces as a service on a true pay per use model. All you need is an Amazon account and your done. Unlike other products, that are offered in similar fashion, we decided not to provide just a bundle of our software packed in an Amazon image (AMI) but instead provide a more tight integration with EC2 that will enable our users to use a single GigaSpaces product for all GigaSpaces versions as well as customize the installation to include their own software packages and application code on the fly. The new CCF provides a true end to end experience that will enable our users to deploy and manage their *entire* application from the load-balancer to the database in one single click.  The latest release of CCF includes support for the new GigaSpaces 7.0 release. Users that are looking to try out the new XAP product can now do that for free without the need to download or install anything.  This will enable you to try out any of the built-in examples, test the high availability features and run benchmarks in just one click. The new version includes lots of additional improvements to the user experience as well as new features and bug fixes. To view a detailed list of the new features click here.

Those that will be visiting JavaOne this week will be able to join the hands-on labs session and experience a full deployment of JEE application using CCF. You will also be able to gain insight from one of our customers, Primatics Financial on how they used CCF to build their Risk Management application as a service – See  full details here

In this post I will try to address some of the most common questions that we often receive when introducing this new service:

Who is using GigaSpaces CCF and Why?

The new release is already used by different customers in the following areas:

ISV’s and Service providers using CCF for SaaS enablement:

Many ISV's are looking for ways to move their standalone applications into a SaaS based offering. In doing so they face a few challenges:

1. Scalability – now their application needs to serve not just the users of a certain customer at a certain location but all customers from all locations. This increases the scalability challenge on their product by an order of magnitude.

2. Continuous high availability – one of the challenges that is imposed by having a shared platform for all customers is that the impact of failure can be catastrophic as failure can affect all of your clients at the same time and becomes public fairly quickly.

3. Portability – There are cases where some customers would still require their application be installed in their local data center for security and latency purposes. Today, this often means you will need to develop two seperate branches of your product to serve these two scenarios. With GigaSpaces you can write your application once and deploy it in the cloud or in your customer data center without having to modify your application and without locking your application to a specific cloud provider.

4. Performance – In developing high performance applications such as Trading, Risk Analysis, or SIP server, keeping the latency at sub msec and throughput at a maximum becomes critical. This is where our XAP middleware has a proven record in running numerious production mission critical applications.

The following are a few public references that chose to use XAP:

- Primatics Financial is using GigaSpaces CCF to build a high performance risk management solution

- Orbyte is using GigaSpaces CCF for building state of the art high performance trading applications that can be offered as service.

Other customers include Online Gaming companies that were looking for a simple and cost effective manner to launch new online games and provide a gaming application platform that enables other gaming providers to host their applications easily in a cloud environment. See further details on our online gaming solution here.


Enterprise customers using CCF to improve time to market of new applications and for large scale testing

According to recent research conducted by Forrester, one of the main requirements driving cloud adoption by the enterprise market is that current IT can’t keep up with the business demands of their organization:

  • Capacity planning is too difficult – The current method of provisioning applications based on estimated peak loads has become either too costly or impossible. Most systems are non linearly scalable which means that knowing how many resources would be needed to meet certain goal is known only through trial and error.
  • Time to market – launching new applications and services often takes months and sometimes year.
  • The business wants a simple and cost effective way to prototype – not all applications make it to production but these initiatives still require the business to invest upfront in the purchase and configuraiton of infrastructure to support these prototypes.
  • Cost saving – There are sets of applications that have fluctuating demand on resources.  Enabling these applications to “lease” a common set of resources only when under heavy demand can lead to huge cost saving and efficiency.

I discussed the enterprises challenges in my previous talks. The most recent one is the one I gave during the CloudSlam event:  Practical Guide to Developing Enterprise Applications on the Cloud- Online Presentation 

A good example of an enterprise customer who is already using our CCF in production is a Large Telco Service Provider that searched for ways to reduce the time to market for launching new scalable web applications. This customer was able to bring a project that previously took more than 6 months just to finish the development and testing to 4 weeks with only 90$ spent on pre-investment.

Large scale testing, performance benchmarks and demo as a service

Users that are not ready to use the cloud to run their production systems can still receive benefits from the cloud for testing and demonstration purposes. There are a few challenges that we tried to address with CCF to make this type of use simpler:

1. Making cloud testing environments identical to your local testing environment – To effectively test your application on the cloud and deploy a production version of your system locally, your application must be at a point where it behaves fairly the same in both environments.

2. Setting up a large scale environment of hundreds of nodes can be fairly complex

3. Trouble shooting failure in such an environment can be a fairly complex and lengthy process.

If your application is already running on our XAP product then porting it to the cloud and running it locally in your production environment will not require any change to your code or configuration. With CCF setting up a large scale environment of hundreds of nodes with the new CCF takes just a few minutes. We designed the CCF to run not only GigaSpaces but other external services which are essential for tests such as JMmeter, Tomcat servers etc. Users can easily configure their own initialization and command scripts when the machine is launched and define which software packages will be installed on their machines on the fly. We added the ability to capture the logs of multiple machines easily and monitor the system and middleware behavior at runtime using a zero installation environment all through the web.

GigaSpaces and our existing customers and partners have already been using CCF to run live demos and recently GigaSpaces started using CCF for our own internal large scale testing purposes. Using the cloud for demo purposes has been extremely useful to show cases that previously were hard to demonstrate such as fail-over and scalability in a real production environment. You can read more on how you can use CCF for running your own demo on the cloud in a similar way without going through all the challenges we had gone through in one of my recent posts: Demo as a Service . Similarly CCF can be ideal for running large scale Benchmarks.

Partner ecosystem -

We designed our CCF in away that would enable us to run not just GigaSpaces components but almost any application such as JEE or Spring based applications. This enables our Solution provider partners to use CCF to offer their specialized solutions and offerings for both GigaSpaces and their other software partners.

System Integrators are also looking for ways to gain a competitive advantage by offering turn key solutions at a significantly shorter time to market and lower initial investment. I tried to outline below some of the public references on these two categories:

  • Solution providers

Real time production monitoringDynatrace offers enhanced integration with CCF for enhanced production monitoring.  Dynatrace users are able to monitor in real time how their application is performing and trace their application bottlenecks in real time – see details on this integration here.

Model Driven Development framework - New Technology/enterprise (NT/e) - GigaSystemBuilder helps Java developers prove applications quickly - in a few days - on a local grid or in the cloud, and then rapidly move on to production development, rather than waste time on finger-trouble or implementation details – see more information here as well as a live demo on their deployment manager solution

  • System integrator

In many cases customers are looking for experts that will enable them to gain the benefit of the cloud and bridge some of the skill-set gaps. AAR is one of our leading system integrators in that field who has already delivered several solutions on top of our GigaSpaces CCF for Enterprise customers. For them, using CCF was a tool to gain a competitive edge and provided their customers the ability to launch new applications in a matter of weeks as opposed to months and at a fraction of the setup (development environment, testing environment, staging environment) and hardware costs that are associated with setting up such an environment with traditional projects.

Here is a quote from one of our System Integrators in the UK that describes their experience in using CCF for the Telco Service provider that I mentioned earlier.

“We use GigaSpaces XAP and CCF to deploy a standard JEE web application on EC2. To address security concerns we kept the business logic outside of the cloud on the customer site. We exposed the business logic through a secured web services channel. We use XAP data grid for maintaining the session high availability and for reducing the latency associated with accessing the remote data center. We used the XAP integrated Jetty container for hosting our web application and through that gained the built-in self healing and auto-scaling provided through the CCF. In this way we were able to deal with potential failure or load without human intervention. From a management and monitoring perspective we were able to leverage the integrated ganglia monitoring and integrated visual representation of different metrics, like cpu/memory/network usage of the entire cluster. We used JMX to bind our custom Mbeans to the integrated ganglia UI. Overall GigaSpaces CCF gave us big boost on utilizing ec2”

I must admit that from our end this was probably the smoothest project we had been engaged with. The reason was probably related to the fact that unlike other projects we had full control over the environment. This enabled us to minimize potential miss configuration errors which is one of the major areas for complexity in setting up such a project. Another benefit was the support. When we faced an issue it was very easy to log-in directly to the machine and fix the problem on the spot without the need to go through a complex process of authorization from the IT Security and without the need to create a reproduction package and ship it over.

The customer was able to deploy a new business application quickly without putting all the initial investment upfront. In this way they gained the freedom to decide at any point if the application should continue, be canned or if they should move it entirely to their local IT.

Looking into the future

The future looks very promising as we continue to work on bringing new customers and businesses towards using CCF and the latest XAP 7.0. We are currently working on supporting the Amazon European datacenter and taking advantage of some of the new XAP 7.0 cluster administration API to enable users to provide thier own custom SLA deployment. At this point, I would like to ask for your specific feedback and wish list for our next major CCF release.

The Need for Speed - Saving Costs By Improving Application Efficiency

In times like these, improving application performance isn't a major focus for most IT organizations. The common perception is that as long as you're meeting the bare minimum demanded by your users, you're okay - anything beyond that is a luxury you can’t afford. Well, I happen to think this perception is dead wrong: these days, you just can’t afford not to invest in high performance. The reason is simple: high performance == higher utilization.

I’ll explain what I mean. If you do something that makes an application run 10 times faster (this is a typical performance boost experienced by GigaSpaces users - and by the way, XAP 7.0 will be even faster), without changing your loads or service levels, then that application will consume 90% less resources. Or in other words, you can consolidate the servers running this application at a ratio of 10:1. The amazing thing is, this isn’t instead of the server consolidation you'll get from vendors like VMware - it comes on top of and in addition to it, because it helps you cram more virtual machines and more applications onto every piece of physical hardware.

A great example of this is an eBay subsidiary, Marktplaats, which has moved its application to XAP and is now expecting to reduce their data center from a few hundred servers to only a handful - the consolidation ratio is a whopping 18:1. Marktplaats says this reduction is largely a result of the huge performance boost they experienced, which was made possible by XAPs In-Memory Data Grid and parallel processing capabilities.

XAP also makes it possible for extreme performance to thrive in unexpected places - one example is an XTP trading platform which, thanks to GigaSpaces XAP, has become SaaS-enabled, a major differentiator for the platforms makers, Orbyte Solutions. Another is our recently-announced joint solution with Mule, the open source ESB, which proves that "high performance SOA" is not an oxymoron :)

May 25, 2009

Interesting talks, and free drinks in JavaOne

Its been two years since I've last visited the JavaOne conference. This year is going to be particularly interesting as its going to be the first major Java event following the Oracle acquisition of Sun.

image I will have a Technical Session on Tuesday titled: Alternative to Google Application Engine for Java™ Technology-Based Applications, where I'm going to outline the difference between the Google and Amazon approach for cloud computing and discuss how we can combine the best of the two approaches. Argyn Kuketayev and Francis de la Cruz from Primatics is going to join me  and present their experience in deploying a risk management application as a service and provide some of the technical details on how they where able to scale-out their application  on the cloud.


Daniel Templeton from Sun Microsystems will have a lab session: PetClinic in the Clouds: Scaling a Classic Enterprise Application In this Hands-on Lab, participants will take a popular Web application (the Spring PetClinic sample application) and modify it so that it can be deployed on the Amazon EC2 cloud computing infrastructure. They will be exposed to using the GigaSpaces platform as a service, in-memory data grid concepts, the OpenSpaces framework, cloud computing concepts, and persistence as a service using Sun's MySQL™ database technology.

image

We are also co-hosting an event Tuesday June 2 at 8PM with our partner Webtide with whom we've done a great integration for Jetty.

Among those who will attend the party will give a chance to win a free book Savvy Guide for cloud computing by Jim Liddle.

*Note: Space is going to be limited so if you want to ensure your place make sure to register on the online registration site that we set for this event.

The list below include all the sessions and labs that I hope to see – any recommendations on other interesting talks would be appreciated.


Session ID

Session Title

Session Type

Speakers and Company

Date/Time

 Venue Room

TS-4605

Enterprise JavaBeans™ 3.1 (EJB™ 3.1) Technology Overview

Technical Session

Kenneth Saks, Sun Microsystems, Inc.; Marina Vatkina, Sun Microsystems, Inc.

Tuesday
June 02
10:50 AM - 11:50 AM

Hall E 134

TS-4308 Architecting Robust Applications for Amazon EC2
Chris Richardson,
Technical Session Chris Richardson Consulting Tuesday
June 02
12:10 PM - 1:10 PM
Esplanade 307-310
TS-4390 Castle in the Clouds: SaaS Enabling JavaServer™ Faces Applications
Technical Session
Lucas Jellema, AMIS
Tuesday
June 02
12:10 PM - 1:10 PM
Esplanade 302
TS-3817 Google App Engine: Java™Technology in the Cloud
Technical Session Toby Reyelts, Google; Max Ross, Google; Don Schwarz, Google
Tuesday
June 02
3:20 PM - 4:20 PM
Hall E 135

TS-5454

Alternative to Google Application Engine for Java™ Technology-Based Applications

Technical Session

Nati Shalom, GigaSpaces

Argyn Kuketayev

Francis de la Cruz

Primatics

Tuesday
June 02
4:40 PM - 5:40 PM

Esplanade 302

LAB-5564BYOL

 PetClinic in the Clouds: Scaling a Classic Enterprise Application

Hands On Lab

Michal Bachorik, Sun Microsystems, Inc.; Shay Hassidim, GigaSpaces; Daniel Templeton, Sun Microsystems, Inc.
Wednesday
June 03

Wednesday

1:35 PM - 3:15 PM

Hall E 132

TS-5214 Java™ Persistence API 2.0: What's New ?
Technical Session Linda DeMichiel, Sun Microsystems, Inc.; Anil Gaur, Sun Microsystems, Inc.
Wednesday
June 03
2:50 PM - 3:50 PM
Hall E 134
BOF-1304
Meet The App Engine (Java™) Team 
BOF Kevin Gibbs, Google; Toby Reyelts, Google; Max Ross, Google; Don Schwarz, Google
Wednesday
June 03
7:45 PM - 8:35 PM
Hall E 135

PAN-5366

Cloud Computing: Show Me the Money

Panel Session

Jeff Barr, Amazon.com; Jeff Collins, Intuit; Chris Fry, Salesforce; Simon Guest, Microsoft; Gregor Hohpe, Google, Inc.; Raghavan Srinivas, Self; Lew Tucker, Sun Microsystems, Inc.

Thursday
June 04
9:30 AM - 10:30 AM

Gateway 102-103

BOF-5392

Grails Integration Strategies

BOF

Dave Klein, Contegix

Thursday
June 04
6:30 PM - 7:20 PM

Esplanade 307-310

TS-5307

Building Next-Generation Web Applications with the Spring 3.0 Web Stack

Technical Session

Keith Donald, SpringSource; Jeremy Grelle, SpringSource

Friday
June 05
12:10 PM - 1:10 PM

Esplanade 307-310

LAB-5960 Storing Data in the Cloud Hands On Lab
Craig Hubbard, Sun Microsystems, Inc.; Chris Kutler, Sun Microsystems, Inc.; Craig McClanahan, Sun Microsystems, Inc.
Thursday
June 04
9:30 AM - 11:10 AM
Hall E 130-131

Tip* - If you want to find your own sessions I would highly recommend using the JavaOne search tool.

Seeyu next week!

May 19, 2009

GigaSpaces based solution makes it to the finalist of Cisco Developer Contest

I was very pleased to read an email from Leonardo, who was the winner of the OpenSpaces Developer Challenge (a worldwide programming contest using the Gigaspaces application server which was held last year), saying that he is now a finalist in the Cisco developer contest. Here's a bit about him and the application he submitted:

About Leonardo

Leonardo worked for several ISPs in various roles as network administrator and java programmer for IT consulting firms, and finally as software architect in high-performance Java EE based projects. He is passionate about parallel programming, distributed computing and more recently semantic web and its applications on software engineering.

Leonardo was the winner of the OpenSpaces Developer Challenge. He enjoys reading about various technologies in the field of computer science. When he is not developing code, he prefers to spend time with family and friends, walk in the park, or watch a movie.

About the application

Resource Management Platform is a proposal to develop an event based platform that leverages AXP, Services Gateway Initiative (OSGI), Jini and JavaSpaces technologies to enable deployment of IP Multimedia Subsystem (IMS) applications based on Session Initiation Protocol (SIP); more specifically, the Call Section Control Function (CSCF) components. It will have admission control mechanisms to manage Call processing.

This solution improves infrastructure manageability for large scale IMS applications. Such a platform will potentially be useful to enable deployment of high-performance, network-based SaaS (Software as a Service) or Cloud Computing solutions at the network edge by leveraging AXP.


You can find the full details about his project here.

Leonardo's project is interesting, because it shows how you can use Space Based Architecture (SBA) for implementing a scalable Telco application and offer it as SaaS application on the cloud.

Interestingly enough, I got another email the week before from Amin Abbaspour, who presented another case study illustrating how you can build a scalable SMS service using SBA, as shown in this diagram:

image 

What the two projects have in common, from an architecture perspective, is that they both represent a highly scalable Event Driven design. The unique thing about Event Driven applications is that they require a combination of messaging, data and service interaction that needs to be tightly orchestrated to meet high performance/low-latency requirements without compromising on consistency, ordering (FIFO) and reliability. This combination of requirements represent one of the hardest challenges in building scalable architectures. Trying to meet this type of challenge in the traditional way by integrating messaging system for event delivery , database or simple caching (like Memcached or TC) for data and a traditional application server for business logic is going to lead to fairly complex architecture. Trying to reach linear scalability and keeping the latency low with so many moving parts is close to impossible. This is what makes SBA such a good fit. The main difference about SBA is that it recognizes there is strong dependency between messaging, data and business logic. The key is to have one shared clustering, high availability and scalability for all three components of the architecture. This makes it possible to reduce the number of moving parts and network hops associated with each business transaction, thereby increasing reliability.

On a personal level, I was very pleased to see that the software we are developing is helping people like Leonardo and Amin to build their own carrier and put themselves in a unique spot in highly competitive market.

Good luck Leonardo and Amin!

References

May 18, 2009

Practical Guide to Developing Enterprise Applications on the Cloud- Online Presentation

According to a recent survey, available skill sets is one of the leading decision factors for organizations in determining which application platform to use, while scalability and availability are next. This reveals one of the main obstacles for bringing enterprise applications to the cloud: How do you take something as disruptive as cloud computing, and bring it to an enterprise environment, without forcing a complete re-write?

In my recent talk at the CloudSlam (online) Conference, I tried to summarize the challenges people face when trying to deploy enterprise applications on the cloud. Try to imagine a scenario where you have an existing JEE system, and your system is under load. At this point you would like to add more machines to accommodate that load. What will happen to your application in this case? Will your application be able to take advantage of the additional capacity? How do you know how many machines should be added to meet the load in the first place?

The answers I usually hear to these questions are fairly consistent. If you’re lucky, you’ll only require configuration changes to be at a point where you can utilize the extra resources. Knowing how many machines to add in order to meet a certain load is yet another challenge. Going through the normal capacity planning process could take weeks with the way systems are currently running.

At the same time, if you’re going to deploy our application in a static manner (reserved instances, static IP, …), it would probably make it simpler to deploy your existing application on the cloud, but it probably won’t make much economical sense. So the question I was trying to answer is how we get from the static application deployment most of us are using today, to a point where we can get end-to-end scaling and elasticity of our application from the load-balancer to the database without going through a complete re-write.

In this presentation, I tried to suggest a set of solutions to overcome the various challenges I mentioned earlier and how to apply them in a gradual manner to avoid huge initial investments and high risk. There were also some interesting questions toward the end, centered specifically around one of the hardest problems – getting the data pieces sorted out in a such a dynamic environment. Those who missed this presentation can now view it online:



Note: The voice quality isn’t that great, plus the desktop sharing on Webex didn’t give a clear indication when my desktop was actually shared, so the beginning of the talk is missing the first two or three slides. If you have questions or want to get a copy of the presentation, just shoot me an email.


During the upcoming JavaOne conference there will a hands-on lab session by Daniel Templeton from Sun Microsystems (PetClinic in the Clouds: Scaling a Classic Enterprise Application) where users can go through the steps and deploy a Pet Clinic application on the cloud. 

In addition to that lab, I'm going to have a Technical Session (Alternative to Google Application Engine for Java™ Technology-Based Applications).

For those that are not going to join the JavaOne event and want to get some hands-on experience on how the steps that i outlined in the presentation can be implemented in real application, I’d recommend looking Pet Clinic demo that is available on Amazon cloud using our cloud framework.

I’m happy to say that the number of customers that are already using this approach through our Enterprise Cloud middleware platform is growing quickly and were seeing more and more application that would traditionally considered as not "cloud compatible", being deployed on the cloud. Below are few of the public references that were mentioned just recently on Jim Liddle's blog:

May 02, 2009

Twitter as a scalability case study – it’s the architecture, stupid!

In May of last year I wrote my first write-up on Twitter scalability, titled Twitter as a scalability case study. Back then, Twitter was still struggling with its Ruby implementation and claimed to be 10000 Percent Faster. During the past few weeks I came across the an interview entitled Twitter jilts Ruby for Scala which described some of the new development that led Twitter to switch some of its Ruby backend development to Scala. I had follow-on discussions with few colleagues ever since WRT to Scala and functional languages in general which led me to write this piece.

I must admit that whenever a scalability discussion becomes a language-choice discussion, I get irritated. It's enough to read a comment such as the one below to see that there is something is missing from the puzzle:

"Today, Payne said, most of the hip dev set codes in Ruby or PHP or Python because they're perceived as "agile" languages.. but also because the deverati grew bored with languages like Java and C++"

After I read the following note, things became a bit clearer:

"By mid-2008, one of these Ruby message queues completely crashed and developers needed two and a half hours to shove the dropped Tweets back through the system. When your game is micro-blogging, that's a lifetime. Then, in his spare time, one developer ported the code to Scala. According to Payne, the Scala queue could process the same message backlog in 20 seconds."

Later, I read Bill Venner’s excellent interview, Twitter on Scala, which i found quite insightful. But after reading the following response on the use of the Actor model in Twitter I felt even more confused:

"..we found that actors weren’t necessarily the ideal concurrency model for all parts of that system. Some parts of the concurrency model of that system are still actor based. For example, it uses a memcache library that Robey wrote, which is actor based. But for other parts we’ve just gone back to a traditional Java threading model. The engineer working on that, John Kalucki, just found it was a little bit easier to test, a bit more predictable. The nice thing was, it took minutes to switch code that was actor based over to something thread based."

Now to be clear, I have nothing against Scala or Ruby or any of the dynamic languages – quite the contrary. I believe that productivity, being agile, and enjoying writing the code are important factors when we make our language choice. At the same time we should remember that these are only some of the factors; other factors which should influence our decision are available tooling, existing skill-set, maturity, performance, etc. I also don’t care that much how Twitter makes their technology choices. What gets me worried is comments such as this one:

“the switch should stand as a lesson to cutting-edge coders everywhere”

When this type of statements gets to the press and gets backed with intelligent stories that justifies these choices, it is clear to me that many "naive" individuals will start following the same choices without understanding the full picture and history that led to the decision, thinking that they can base their decision on “proven success”. Looking at the history could evil a fairly different picture.

What I found missing in the entire discussion both here in Twitter case and in many of the other web2.0 scalabilty architecture stories is something that will indicate that there is some consistent methodology behind the  architectural choices. For example, most of the limitations that were mentioned about Ruby’s threading model, as well as its memory utilization issues, were well known before Twitter chose Ruby as their core language. Why did it take such a long time to realize that you can’t come up with scalable architecture without addressing those limitations? I could argue the exact same thing about Digg and others.

What i would recommend is to look at Brian Zimmer’s summary on Scalability Worst Practices, in particular the “Golden Hammer” point:

"The Golden Hammer refers to the old adage: if all you have is a hammer, everything looks like a nail. Many developers fall prey to the idea of using only one technology – the cost of this is having to build and maintain an infrastructure in the chosen technology which may be readily available in another technology which is more suited to the specific problem area's functionality or abstractions. Forcing a particular technology to work in ways it was not intended is sometimes counter-productive."

The lesson in our specific Twitter case is that language choice should be used as a means for implementing a given architecture and not the other way around. We may find that implementing various parts of our architecture require different technologies and languages, and that’s perfectly fine, as long it follows that order. In Twitter’s case, choosing Ruby for the front-end and Scala and Java for the heavy stuff sounds reasonable, only that I would expect them to get to this realization much sooner.

Final words

In this blog I pointed my comments specifically at Twitter, however Twitter is only an example. Many Web 2.0 sites developed their scalability architecture in a similar trial-and-error approach, which, as we can learn by their histories, tends to be very painful and costly. And yet, building a scalable Web 2.0 application such as Twitter shouldn’t be rocket science. I’m certain that if you follow the right design principles and learn from other proven scalability patterns, you can avoid a large part of the painful “trial and error” experience before coming up with the right solution. Making the right build vs buy decisions can make a lot of difference. In one of my recent posts, Designing a Scalable Twitter, I tried to suggest a methodology for coming up with scalable Twitter architecture. Like anything in life, I’m sure that there is more then one possible solution and more then one language that could be used to implement our solution. In many cases you will find that even though language and platforms could vary between different implementations, the architecture principles remain pretty much the same.

To sum that up, I would say that scalability is first and foremost an architecture choice; different languages can make the implementation of a certain architecture simpler, but language by itself doesn’t make any application more scalable. Different parts of an application may require different set of languages, and choosing the right langrage should also take into consideration existing skillset, maturity, existing development tools and best practices – and obviously, the application’s performance and scalability characteristics.

April 14, 2009

Designing a Scalable Twitter

Guy Nirpaz, Uri Cohen and Shay Banon came up with an interesting exercise as part of the recent partner training that took place at the GigaSpaces office. In this exercise, the students were asked to come up with a scalable design for Twitter, using Space-Based Architecture.

There are some interesting scalability lessons from this exercise, which are applicable to anyone looking to implement new-style real-time web applications such as the ones used for social networking.

In this post I'll  try to summarize the main patterns to put into place and considerations to make when designing such a scalable architecture.

Background:

For those of you who are not yet familiar with the service, Twitter is sort of a SMS-service meets discussion board.  You can post short messages (up to 140 characters) that can be shared with a group of subscribers that are referred to as "followers". The main difference between twitter and other messaging applications is that both SMS and Instant Messaging (IM) applications were designed primarily for one-on-one communications whereis Twitter was designed primarily for broadcast communications (publish/subscribe, or pub/sub). Another aspect that is special about Twitter is that by default anyone can follow anyone else. In other words, it was designed for open communications, not private, as were IM and SMS.

What are Twitter's scalability challenges?

1. Sending a tweet (a message on Twitter is known as a 'tweet') -– The challenge is how to handle an ever-growing volume of tweets and re-tweets and responses that can lead to a viral "message storm"

2. Reading tweets – The challenge is how to handle a large number of concurrent users that continually “listen” for tweets from users (or topics) they follow.

Designing A Scalable Twitter

Choosing the right scalability patterns

Almost every challenge in software architecture has its roots in one of the existing patterns. So the simplest course is to start by looking for those patterns, and choosing the right patterns to scale the application. Looking at many other scalable architectures, we'll begin with a partitioning pattern as the core design principle. By partitioning our Twitter-like application we'll spread the load across a cluster of servers and scale by simply adding more servers (i.e., partitions).  Another important architectural observation about Twitter is that it doesn’t fit into the classic database-centric design that most web applications do. On the flip side, it doesn’t fit well with a messaging-centric design (pub/sub) either. It is a combination of the two.

A pattern that is suitable for this type of collaborative messaging is known as a blackboard pattern.  In our design, we will use those two design patterns -- partitioning and blackboard -- as the foundation for our scalable Twitter application. With the foundation in place, let’s list the requirements and examine how these patterns can be used to scale the app.

Scalability Requirements

We'll assume a relatively extreme scaling requirement:

  • Tweet Volume: 10 billion tweets per day
  • Tweet Storage: 100 Gigabytes per day (with 10:1 compression)

Additional assumptions:

  • Tweets are limited to 140 characters
  • Tweets are immutable, i.e., there are no updates, only inserts
  • Twitter limits client applications to 70 requests per hour

Now that we have the foundational patterns and clear requirements, we can design the architecture. We'll start first with the blackboard system.

Using an In-Memory Data Grid (IMDG) as a Blackboard System

The are several approaches to building a blackboard system. To maximize performance and scalability, we'll store the data in memory, thus avoiding disk I/O, which is often the main cause for contention. For years, Java has provided a model for designing blackboard systems known as JavaSpaces. More recently, distributed caching has become popular and can provide similar capabilities to those of JavaSpaces. Let's examine two popular distributed caching approaches for our blackboard system:

  1. Simple read-mostly caching using memcached
  2. Read/write caching, also known as an In-Memory Data Grid (IMDG)

Choosing between memcached and an IMDG

Memcached enables us to to store the data (tweets) in a distributed memory set and read it in a scalable fashion. Having said that, be aware that memcached is not transactionally-safe and is not designed for reliability (i.e., it doesn’t support fail-over and high availability). That means that if we use memcached or something similar, we will have to use a database as the back-end. Every tweet posted will have to be written to both memcached and the database in a synchronous fashion to ensure that no tweet will be lost. This approach may be good enough for scaling read access, however, for writes and updates it offers limited scalability.

Unlike memcached, which was designed for simple read-mostly caching, In-Memory Data Grids  are designed for handling a read/write scenario, and can therefore act as the system-of-record for both write and read operations. We can still use a database for long-term persistence, but because the IMDG maintains its reliability purely in memory, we can write and update the database asynchronously and avoid hitting the database bottleneck.

Todd Hoff, author of highscalability.com wrote an interesting summary that covers the different products in this space in a recent post:  Are Cloud Based Memory Architectures the Next Big Thing?

Todd provide a clear explanation of how an IMDG works (using GigaSpaces):


Nati blog 1 (2)
Natiblog 2 (2)  

  • A POJO (Plain Old Java Object) is written through a proxy using a hash-based data routing mechanism to be stored in a partition on a Processing Unit. Attributes of the object are used as a key. This is straightforward hash based partitioning like you would use with memcached.
  • You are operating through GigaSpace's framework/container so they can automatically handle things like messaging, sending change events, replication, failover, master-worker pattern, map-reduce, transactions, parallel processing, parallel query processing, and write-behind to databases.
  • Scaling is accomplished by dividing your objects into more partitions and assigning the partitions to Processing Unit instances which run on nodes-- a scale-out strategy. Objects are kept in RAM and the objects contain both state and behavior. A Service Grid component supports the dynamic creation and termination of Processing Units.

Back to our Twitter app: Given the scalability requirements, we will need to scale both reads and writes, and therefore, an IMDG is a more suitable approach to implementing the blackboard system.

Now let’s examine how the use of an IMDG as the blackboard system enables us to scale both sending and reading tweets. Let's start by designing the partitioned cluster.

Designing a partition architecture

One of the main considerations in designing a partition cluster of any kind is determining the partition key, such as a Customer ID in a CRM application or a Trade ID in a trading application. At first glance, it sounds like a trivial decision, but choosing the right partitioning key requires a deep understanding of the application usage patterns and data model.  In the case of Twitter, we could choose to partition the application by the data-type, the user, the tweet itself or the followers. Our first goal is selecting a key that will that will be granular enough to enable scaling the application just by adding more partitions, while making sure that we don't end up with a key that is too fine-grained -- making it sub-optimal for querying purposes.

If we use the timestamp key, for example, our application will be optimized for “inserts” (writes), however, even a simple query such “retrieve the tweets of a certain user” will force us to execute an aggregated query against all partitions. Alternatively, if we partition the data based on user-id, we'll be able to easily spread the load from different users across partitions. Retrieving the tweets of a certain user is going to be resolved in one call to a single partition. We may encounter a problem if a single user generates a significant higher load than average, however, in the case of Twitter, we can assume that this is not very likely. Partitioning by user-id is a good compromise.

Data capacity analysis

With such extreme requirements it is clear that storing all tweets in memory is going to require huge memory capacity. Very quickly this will become economically prohibitive, so we need to devise a scheme in which the IMDG acts as a buffer for most of the load on the system, and then offloads the data and queries to an underlying persistent storage.  In our Twitter example, it is fair to assume that most real-time queries (those that require fast access to the data) will be resolved in data from the last hour or 24 hours. Queries that require older data will need to hit the database for the initial call. However, subsequent access to fetch new updates should be resolved purely in-memory.

Using this approach, we'll need about 10 servers, each holding 10GB of data in memory to accommodate 24 hours of activity. If we also want to back up the data in memory, we will need double the amount of servers.

Choosing the right eviction policy

It's reasonable to assume that recent data is accessed most and older data is rarely used. To ensure that we get the maximum hit ratio on our memory front-end, let's choose a time-based eviction policy, which always holds the most recent updates in memory. When we will reach our memory capacity limit the oldest data will automatically get evicted from memory. The actual window of time in which we will be able to keep in memory is obviously dependent on the size of the cluster. With an IMDG implementation all tweets are stored in a persistent storage, which means that when tweets are evicted they are not deleted from the system.

Scaling tweet writes:

If we select user-id as the partitioning key, each user tweet will be sent to a specific partition. Multiple users may be routed to the same partition. Usually the algorithm to determine which partition fits a certain user is something like:

routing-key.hashCode() % #of partitions

In GigaSpaces, this is done by marking the routing attribute of our tweet class with an @SpaceRouting annotation.

The web front-end application will call space.write( new Tweet(..),..)  to send the tweets. This way there is nothing in our web client code that exposes the fact that the underlying implementation interacts with a cluster of partitions (spaces in GigaSpaces). Those details are abstracted within the space proxy. When the write method is called on the space proxy it parses the field that matches @SpaceRouting from our Tweet() object and uses this field value to calculate the partition it belongs to. It then uses that value to route the Tweet(..) object to the appropriate partition.

With this approach, the web application can be written in a very simple way and can interact with the entire cluster as if it was a single server.

Natiblog 3

The data from the memory partitions gets stored asynchronously into a persistent storage. The persistent storage could be a database, but it could also be other things, such as an index search engine based on Compass/Lucene.

Scaling tweet reads:

To those familiar with messaging system, at first glance Twitter looks like a classic publish subscribe application. A closer look, however, reveals that any attempt to implement Twitter with something like a JMS message queue is going to fail in achieving a scalable system. This is especially true if you consider that the system needs to maintain a durable queue for each user. That could easily lead to a scenario in which each tweet is published to thousands of subscribers and every re-tweet can potentially lead to a "message storm".

As I discuss above, the right way to think about this type of application is as a blackboard pattern, just as a blackboard (or these days, a whiteboard) is used by a group of people (followers, in the case of Twitter) to share information and collaborate. When someone writes something on the board, everyone sees it and can choose to react. Unlike messaging (take email for example), we don’t need to send separate messages to each subscriber. Instead everyone is looking at the same board. Everything is also copied from the board to paper. When the board runs out of space, we erase it. And we can always page through the paper copy to access the board history. 

In Twitter, this means that each follower that follows a group of people is basically polling for messages posted by those users from the last time he read them. To make things more tangible we can express this type of query with the following SQL syntax:

SELECT * FROM Post WHERE UserID=<id> AND PostedOn > <from date>.

The <from date> will normally be the last few minutes, if we're constantly looking for new messages.

But there's a caveat. Remember that we partitioned the application by user-id? This means that each user's tweets are stored in a separate partition. How can we read all users' posts? If we poll for each user individually, we will end up with a lot of network calls. The simplest approach would be to execute one call that looks for ALL the users we're following and look for updates (new tweets) from those users. The pattern we'll use to perform such this task is mapreduce. One way to do that with GigaSpaces is through the distributed task API:

Nati blog 4

The distributed task API is a modern version of the stored procedure. The following snippet shows what such a call would look like:

AsyncFuture<Long> future = gigaSpace.execute(new GetTweetsUpdates());
long result = future.get(); // result will be the number of primary spaces

The GetTweetsUpdates() class contains code that will be injected in each partition and will enable us to look for updates from the users we follow in a single call. Because the call runs in-process, and because the data is stored in-memory, executing such a task is extremely fast compared with the equivalent with database and stored procedure operations. Execution is aggregated to the caller implicitly. The caller can use a reducer to aggregate the results into a single result object.

Scaling the web front-end

Nothing really new here. We'll use a classic web front-end, which is comprised of a load-balancer and a cluster of web servers that act as a front end to our IMDG instances. The web application will use a single cluster-aware IMDG proxy to send new tweet posts. The IMDG proxy will be responsible for mapping the tweet with the actual partition that hosting the tweet. That logic is kept completely out of the application code. This allows us to keep our web tier clean and simple.

Keeping the web layer stateless to avoid session stickiness

One common pattern for keeping the web tier scalable is to use a Shared-Nothing Architecture, which basically means that the web tier will be stateless. This requires keeping the user session state external to the web-tier. As previously demonstrated, the IMDG can be used as high-performance, scalable data store for maintaining shared session state information. This allows us to avoid session stickiness and to scale the web tier without being locked in to a specific server throughout the entire session, in case the server is over-loaded.

For more information on how to scale the web tier, as well as other important capabilities such as self-healing and auto-scaling, see the following tutorial: Scaling Your Web Application.

Making it simple and cost-effective using cloud computing

Twitter is yet another example for a situation in which system load is highly variable and the difference between average load and peak load can be quite significant. In such cases, provisioning our system can be fairly hard and costly. This is where cloud computing and SLA-driven deployments can help us scale on demand and pay only for what we use.

Once we figured out a way to partition the application, it's going to be much simpler to package the application into self-sufficient units (referred to in GigaSpaces as processing-units) and scale the application simply by adding or removing these units on demand. You can learn more about this here

Final words

Scaling a real-time web application such as Twitter or Facebook introduces unique challenges that are are quite different from those of a "classic" database-centric application. The most profound difference is the fact that unlike with traditional sites, Twitter is a heavy read/write application, and not read-mostly. This seemingly minor difference can break most existing models for web application scalability. Using a combination of memcached + MySQL is not going to cut it for this type of application. 

The good news is that with the right patterns and set of tools, building a scalable architecture that meets such challenges isn’t that difficult.  There are already plenty of success stories that demonstrate that, such as the following example from highscalability.com: Handle 1 Billion Events Per Day Using a Memory Grid

The proposed architecture is by no means perfect and can be further optimized to meet even better performance and latency, but that will come at the cost of simplicity. I believe that the proposed architecture should get you pretty far as-is. Avoid going through more advanced optimizations until the point they are an absolute must.


References

My Photo

Twitter Updates

    follow me on Twitter