Nati Shalom's Blog

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

How Map/Reduce and the Cloud are Affecting Analytics Applications

Two weeks ago I gave a presentation at the Sun HPCW workshop in beautiful Regensburg, Germany. It was a great chance to meet many of the folks from the Sun Grid Engine team face to face.  It was also a great chance to take a step back and look at the evolution of one of the interesting areas in the industry - analytics. Even though this is one of the more technically challenging areas, I find it odd that relatively little has been discussed about this within the developers community. 

In this post I will try to provide an overview of the different approaches for running analytic applications. I will also examine how patterns like Map/Reduce and a new environment such as the Cloud, drive a new approach that is significantly simpler and more cost effective than many of the existing approaches. There are other models for analytics such as stream based processing or CEP that i chose not to cover in this post as they deserve represent a new paradigm that deserves a separate discussion.

What is Real-Time Analytics (a layman’s definition)?

Here is the simplest and shortest definition I could come up with (probably not the most accurate one, but good still good enough):

· Analytics: The process of manipulating raw data (a.k.a data mining) into a meaningful report.

· Real-Time Analytics: The ability to generate the desired report in near real time.

Analytics in the Finance Industry

It is interesting to note that financial applications within organizations are organized based on the “speed” with which they need to react to the market. The Front-Office for example, would be the trading/market data application that needs to react to the market event in a fraction of a millisecond. The Middle and Back-office is the part of the organization that looks at an aggregated view of the market and tries to draw various trends, based on the current market and history. The algorithm that is used to process that data and generate the required report is referred to as Monte Carlo simulation.   A typical set of reports generated in that model would be P&L (Profit & Loss), VAR (Value at Risk), Reconciliation etc.

In previous years, those reports were generated on a post trade basis, i.e. at the end of the trading day, the data from the current day was aggregated into files or a database, and during the night there was a set of services that processed the data and generated the desired reports. That meant that if there was an event during the trading day that could affect the portfolio, it would only be taken into account the next day. At the same time that the volume of data increased, the window of time to process that data (during the night) became shorter due to globalization (time zone differences shortened the “night”). This led to scenarios where simulation accuracy was limited by these margins.

Analytics Becomes Mainstream

In general you can categorize any application that looks at an aggregated view of a stream of data, as an analytic application. Fraud detection would probably be another good example of analytics, since it tries to continuously monitor user actions in order to determine whether their action is part of a fraud or not. Social networks, Ecommerce and even personalization, such as search engine personalization have become more mainstream. With that, the demand to process vast amounts of data to produce various market trends, user behavior, fraud behavior etc. becomes not just useful, but critical to the success of the business. A search engine would be a good example of that – the quality of the search results is a direct function of how well you are able to aggregate content and user search patterns, and apply it to the way you produce the search results.

The Evolution of Analytic Application Architecture

To make things simple, I will start with a very basic view of the analytical application architecture, which represents the two most common architectures:

Phase 1: First-Generation Analytical Applications

Phase 1-1: Excel spreadsheet

Everyone uses Excel to generate various reports, so the use of Excel and spreadsheets are probably the most common and popular way of generating reports. The way this is done is by loading the data from a file, and then using Excel to generate the graphs and various views of that data.

Phase 1-2: High-End ETL process

Excel works well for small data sets, but for large data sets a batch process is used that is based on ETL (Extract Transfer and Load). It represents a batch process in which the operational data from the production sites is normally aggregated into some sort of data warehouse. In the process of doing so, the data is stored in a format that is optimized for generating reports. The process normally involves extracting the data, transferring it into an optimized format, and loading it into the warehouse for further processing. The ETL process normally happens once per data source. The data processing (a.k.a data mining) is where most of the heavy duty work takes place. In many cases, once the relevant reports and analysis have been produced, the data becomes obsolete and at this point it is common to clear it and store it in a long term archive.

The Need for Real-Time Analytics

You don’t need to be a genius to realize that that if all reports could be generated in real time, this would probably be the ideal world. The only reason this is not done, is purely technical. No way had been found to aggregate all this vast amount of data and process it in real time. Traditional databases had to be tuned completely differently, to deal with real time transaction processing and complex analytics (type of locking, data structure, transactional semantics etc.).

“Real-time data … requires a continuous flow of data, and traditional enterprise data warehouse deployments introduce slow data delivery to BI environments” (Forrester)

The challenge became even worse as the volume of data increased exponentially, and at the same time there was a continuous pressure by businesses to process even more complex analysis at an even lower cost. It was clear that addressing those contracting requirements couldn’t be achieved by just “throwing more hardware” into the problem. That led to the next phase of analytic application architecture, i.e. a grid-based analytic system.

Phase 2: Second-Generation Grid-Based Analytical Applications

Phase 2-1: Speeding up Analytics via Parallel Processing

Phase 2 was not yet trying to get into real-time analytics but was mainly trying to find ways to speed up the process of analytics. The majority of the focus in that area was on the batch processing that in some cases could take days or hours. By introducing grid computing, the processing time could be brought down from days to hours, and from hours to minutes.

With grid computing, the way that processing time is speeded up, is simply by spreading the processing into a large pool of commodity hardware. By splitting a linear process into small chunks that can be processed in parallel, the processing time can be speeded up quite significantly.

Phase 2-2: Adding an In-Memory Data Grid

As the time to process the data went down from hours to minutes, other factors that were previously considered insignificant became much more significant. It became clear that to speed up the analytics beyond the current level, closer to real time, it was necessary to consider ways to speed up the load and data access time, because they became more significant relative to the processing time.

“Real time drives database virtualization - Database virtualization will enable real-time business intelligence through a memory grid that permeates an infrastructure at all levels" (Forrester)

This approach became fairly common during the past few years. There is more information about this approach in one of my previous posts:  Bringing Data Awareness to the Grid

Challenges with Grid-Based Analytical Systems

While technically the combination of Compute and Data Grid were able to bring the time it takes to perform even very complex analytics to near real-time, the adoption of that solution was still marginal. One of the main reasons that this solution was not widely adopted was cost and complexity. Only a few organizations could afford building and maintaining large data centers. The skill set for running and building applications that could run in those environments was fairly rare, and therefore the labor cost was also rather high.

Phase 3: Next-Generation Public/Private Cloud-Based Analytics

The cloud makes a grid-based system affordable. So with the cloud it is possible to reduce the hardware cost, and more importantly it avoids the huge initial investment. The fact that the cloud is more widely adopted also reduces the labor cost and ramp-up time required to build such systems.

Having said that, the fact that anyone, even the smallest startup can have a large data center at a distance, reduces not just the initial investment and the direct hardware cost, but opens up a new set of opportunities for building such applications. For example it is possible to outsource the entire analytic process to someone else and use it as a service.

This was the case with Primatics Financial that provides risk analysis for mortgage portfolios as a service.

The Cloud and Map/Reduce Make Analytics Simple

With grid-based systems that run in the data center, the pool of machines is fixed. One of the main roles of the grid scheduler is to manage the pool of physical resources which belong to that pool. The cloud on the other hand already takes care of most of the resource scheduling part. Virtualization makes it possible to get the isolation required to maintain the separation between multiple jobs. If these two factors can be leveraged, a simpler model can be developed.

One of the ways to simplify the analytic application is to simplify the way that parallelization is handled. Since most of the resource scheduling is dealt with through the Cloud/Virtualization layer, a simple pattern such as Master/Worker, Map/Reduce can be used to parallelize, i.e. to distribute the application across the available machines.

You can read more about the Map/Reduce, Master/Worker approach here, and how it compares with a grid-based system.

Reducing the Number of Moving Parts

Another way to simplify and optimize the application is to reduce the number of moving parts.

Rather than having a different set of services to load the data, hold it in memory, and yet another set of services that computes the analytics, it is possible to bundle the loader, the data and the processing services into one application unit, and have multiple instances of those units running to handle the scale. In this case, each unit handles the entire life cycle (the loading, processing etc.) of the data analytics food chain.

SaaS Enablement through Multitenancy Support

One of the core elements that is required to provide the analytics application as a service is multitenancy. Multitenancy is the ability to run the multiple analytics processes on a shared infrastructure as if they were running on dedicated resources. To add multitenancy to our service, the life cycle (Start, Stop etc.) of each of the analytics needs to be managed independently of the other analytical processes that are running over the network.

There are two possible layers of isolation that can be chosen:

Service-Level Isolation

Service-level isolation provides isolation at the application component level. You can deploy multiple services on the same process/machine; you can run different versions of that same service in the shared environment and maintain them in complete isolation to other services that are running in the same environment.

Since all services share the same machine, they can potentially access shared resources such as the file system, etc. It is also harder to control things like CPU utilization and memory allocation on a per service level.

Machine-Level Isolation

Machine-level isolation is provided through the use of a virtual machine. In this case, each application or group of applications runs on separate virtual machines. Virtual machines provide an additional level of isolation that includes the actual CPU, Disk and Memory resources.

Combining Service-Level and Machine-Level Isolation

Both service-level and machine-level isolation are actually complementary approaches, as each one of them provide a different level of isolation and application packaging granularity. Service-level virtualization tends to be more fine grained, but at the same time provides a lower level of isolation. Therefore, it makes sense to use it in areas where real CPU, Memory and file system isolation are not that critical, but getting better utilization from the existing machines is more important, and vice-versa.

In the diagram above, you can choose to run everything from the same organization at a service level of isolation, and use a VM level of isolation between different customers to ensure that each customer’s application is running in a completely private set of VMs.

Private/Public Cloud

With the introduction of private cloud providers such as VMware with their Vcloud and Vsphere approach, and open source alternatives such as Eucalyptus, you can choose to run the same solution in a hosted environment or locally. You can also use a hybrid approach where some of the processing runs in your local environment and the rest in the hosted environment. Things like the Amazon Virtual Private cloud make such a hybrid deployment model fairly simple, as the local and hosted machine are accessed in the exact same way.

Final Words

The continued demand for processing a growing amount of data in real-time and at a lower cost, drives continuous innovation in the way data is processed.  Grid-based analytics (Compute and Data) addressed the demand for scaling, and also reduced processing time quite significantly, but at a relatively high cost and complexity. The demand for real time analytics by mainstream applications in the social networking media created a drive for a simpler and less costly solution. Cloud computing enables the cost of hardware to be reduced, but at the same time opens up new opportunities for making the analytic processing significantly simpler. You can rely on the cloud for scheduling and managing your resources, and on virtualization for multitenancy. With that, a simplified parallel processing approach, such as Map/Reduce, can be used to handle the parallelization of the application.

“it's quite clear that the business-intelligence/data-warehouse industry is moving toward a new paradigm wherein the optimal data-persistence model will be provisioned automatically to each node based on its deployment role -- and in which data will be written to whatever blend of virtualized memory and disk best suits applications' real-time requirements." (Forrester)

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):

 

• 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.

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:

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

• Twitter getting bigger - Scaling becomes more of an issue

• Memory-Based Architectures and the Cloud

• Handle 1 Billion Events Per Day Using a Memory Grid

• Are Cloud-Based Memory Architectures the Next Big Thing?

• Scaling Your Web Application

• Persistence on the Cloud – How and Why?

• Scaling twitter presentation by Guy Nirpaz

Reducing latency with Sun Real Time JVM

Frederic Pariente  Engineering Manager at Sun Microsystems posted an interesting summary of a case study with GigaSpaces on Sun blog: Gigaspaces curbs latency outliers with Java Real Time

In the context of a customer proof-of-concept this summer and in the light of the 2.0 release of Java Real Time System --JRTS 1.0 had the bad prerequisite of source code changes--, Gigaspaces revisited the opportunity for Java Real Time to serve the low-latency requirements of trading applications. Gigaspaces XAP 6.5, Solaris 10 and both Java 5.0 standard and real-time JVMs were used for the benchmark. The test scenario included a trade matching engine and multiple clients injecting messages at extreme speed. The success criteria was to get guaranteed latency per message under 10 msec, with no code modification to the matching engine"

"..The first lesson learned was that msec latency was achievable with the standard JVM, through some advanced tuning of the JVM command-line options. While the customer had reported application freezes up to 20 sec during garbage collection under heavy load --he was running the JVM with no particular flags, unfortunately default JVM options optimize for throughput--, latencies could be brought down to milliseconds by switching to the Concurrent Garbarge Collector"

"..The second lesson learned was that the number of outliers can be reduced by an order of magnitude by using the real-time JVM. At a small cost in terms of application throughput --lower-- and CPU usage --higher-- of course"

You can see the full detailed benchmark and JVM option in the original post.

How Real Time JVM works?

For those who are not faimiliare with Real Time JVM, Fredric points to a very detailed presentation on his post Java Real Time for latency-critical banking applications  which I'd recommend looking at. I took one slide from the presentation which i found useful for understanding the general concept behind Real Time VM.

As could be seen in the above diagram the RT-JVM introduces new type of threads "Critical RT threads". It makes sure that GC will not run while those thread are running and in that case provide better predictable behavior.

Other references:

Latency is Everywhere and it Costs You Sales - How to Crush it - My Take