Nati Shalom's Blog

NoCAP – Part III – GigaSpaces clustering explained..

In part-I of this post I tried to suggest a scalable architecture that can  deal all three properties of CAP. Part II goes back to the definition of Availability and Partition tolerance and the different views on that regard. In this post I would like to elaborate more specifically on how the suggested architecture from part-I works in the context of all three CAP properties.

Recap from previous posts

According to the CAP theorem, one should choose two out of the three CAP properties: consistency (C), availability (A), and partition tolerance (P). Some suggest to choose AP and compromise on consistency. Others like Stonebreaker suggest CA as a better set of tradeoffs:

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

My suggestion followed a similar line of thought as presented by Dr. Stonebreaker, with the addition that I argued that we don’t necessarily need to give away partition tolerance completely when we choose CA.

I suggested an alternative approach where, instead of looking at each of the CAP properties in absolute terms and selecting only two out of the three, we could use a more relaxed version where we could apply various degrees of all three and compromise on the degree in which we apply each property based on the business requirements of our applications.

We will address the most likelihood scenario in which we expect to experience failure or partition tolerance and compromise in areas which are less likely to happen.

In Part I, I used one of the common GigaSpaces clustering topologies as a reference to that model. In the next section I’ll explain how that specific topology applies to all three CAP properties.

GigaSpaces clustering explained..

The following diagram taken from part-I of this series of posts describes a clustering topology that was designed to deal with scalability, throughput as well as consistency, availability and partition-tolerance.

How it works (Quoting from the original post…)

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

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

Now let’s examine how we can apply the various CAP properties into that architecture.

1. Consistency 

• Consistency under concurrent updates. To ensure consistency  in the case of concurrent updates on the same data record each individual record is mapped to a single logical-partition at each given point in time.  To ensure scalability, different records of the same logical table are written to multiple partitions in parallel as described in the diagram above. Each partition support the various locking semantics (pessimistic, optimistic (versioning), dirty-read) etc.. to control the concurrent access of the same record within the context of a single partition.
• Consistency between two or more replicas. To ensure the continuous high availability we keep one or more copies of our data. In asynchronous replication, we may end up with scenarios where read and write operations would hit two different nodes at the same time and end up reading two separate versions of that same data. There are various algorithms that were developed to handle that situation. In our case we chose to avoid getting into that situation in the first place through the use of synchronous backup.  The performance overhead of the synchronous replication is fixed and is not proportional to the size of the cluster (each partition replicates data only to its backup replica). The replication to the database is kept asynchronous to reduce the overhead of writing to disk.
• Transaction consistency. Single operations or groups of operations can be executed under transactions. This ensures the ACID properties. Transactions can be made local to each partition. In this case, they will bound to the scope of a single partition and would be highly optimized in terms of performance. Transaction can also span between nodes (in this case the overhead is obviously going to be higher).
• Ordering – All operations are ordered based on the time they were written. This is specifically relevant to ensure the consistency between the in-memory cluster and the long-term storage which is being updated asynchronously.

2. Availability 

• Primary failure. We always keep one or more replica nodes as a hot backup. The hot backup will take over immediately in case a primary node fails. The hot backup nodes uses synchronous replication to ensure no data loss before fail-over took place.
• Backup failure. When a backup node fails, the primary continues to serve requests and log the operation in a redo-log.  In parallel, a new backup is being provisioned on demand to take over from the failed one. That process involves a provisioning process (in which a new backup is created) and a recovery process (during which the backup gets its state).
• Failure of multiple nodes.  To increase availability, some of the NoSQL variants suggest at least three or more replicas per partition. In this way, we can handle simultaneous failure of multiple nodes. That obviously comes with a huge overhead – for each terabyte of data, we would have two terabytes of redundant information for backup purposes, and - equally important - we would also have the consistent overhead of keeping all of them up to date. An alternative approach is to use on-demand backups. On-demand backups are provisioned automatically as soon as one of the nodes fail. If spare capacity exists within the current pool of machines the backup will be provisioned into an alternate machine within the existing pool – this process can take a few seconds (depending on the amount of data per partition). If no machine is available, it will start a completely new machine.A new backup will be providioned into that new machine. The process of starting a new machine with its backup can take few minutes.  As soon the node starts it will first use a primary election protocol to find the master node within its group and only then it boots up. The startup process include a recovery stage in which the node recovers its state from either the master node or the available replicas. The source node will also store all the updates since the recovery started in a redo log and would replay all updates to fill in the gap since the node started its recovery process.
• Client failure.  Clients use a cluster-aware proxy to communicate with the cluster. The smart proxy ensures that a write or read operation is always routed to one of the available partitions. The routing happens implicitly thus the client is not exposed to a fail-over scenario.
• Discovery protocol. The GigaSpaces cluster discovery mechanism is based on the Jini specification. Services use the discovery protocol to find nodes within the cluster and share cluster state amongst all nodes.

3. Partition Tolerance

• Network partition between primary and backup. When the connection between two nodes fails, the primary node logs all the transactions into a fifo queue known as the redo-log. As soon as the communication gets re-established all the data gets replayed to the backup. If the backup fails completely, the system will start a new instance as described above.
• Network partition with the long term persistency. If the communication with the long-term persistency datastore fails, the replica will log all the operations till the connection gets re-established. The log is also replicated to a backup node to ensure that the data won’t be lost in case the primary partition fails before the data was successfully committed to the long-term storage. 
• Network partition between two or more data center sites. Most people referred to scenarios where two sites can live in two separate locations and continue to work independently in a case of network partition as THE reference scenario for partition tolerance. It is important to note that there are two class of multi-site deployments that are fundamentally different as it relates to the network partition:
  • Disaster recovery site – is often located in close geographical proximity to the primary site and is backed by high  bandwidth and redundant  network. 
  • Geographically distributed sites over internet WAN – in this case we have multiple sites that are spread over the globe. Unlike disaster recovery sites those sites tend to operate under lower SLA’s and significantly higher latency.

The recommended approach for dealing with multisite network partition would be fairly different in each of the categories that are mentioned above:

Disaster recover site. Disaster recovery sites are very much like any node in a local network but often live in different network segments and with higher latency than local networks. Nodes within a cluster can be tagged with a zones tag to mark their data-center affinity. The system can use this information to automatically provision primary and backup nodes between the two sites. It will use the zone tag to ensure that primary and backup are always spread between two data centers.

• Consistency & Availability - The consistency and availability mode would be just the same as LAN based deployment as described above but the performance and throughput per partition would be lower due to the higher latency associated with the synchronous replication.
• Partition tolerance – The system will continue to function even in a case where entire site fails or become disconnected. The system will continue to work through the available site.  The system will also rebalance itself as soon as the communication between both sites gets re-established in order that the load will be evenly distributed between the sites.

Geographically distributed sites over internet WAN. In this scenario, nodes are spread over internet connections where the SLAs are lower and latency is significantly higher. In that case, it would be impractical to treat all of the nodes as a single cluster as in the previous case. It would be more practical to use a federated cluster deployment (a cluster of clusters) where we will use asynchronous replication to synchronize the multiple sites and therefore avoid the extreme latency overhead. In this mode, we can’t achieve all three CAP properties. In most cases it would be more common to choose AP over CA. Having said that even when we choose AP we don’t necessarily need to give away consistency completely. Here is a suggested architecture that can provides reasonable degree of consistency with slight compromise on absolute availability and partition tolerance:

•  
  • Consistency: Each site is the sole owner of its data – i.e. all updates on the data that belong to this site needs to be delegated to this site. In more extreme scenario we can define a master site which will own all the updates to the data by all other satellite sites. In that case we can ensure consistency and read availability over write availability.
  • Partition tolerance: All sites use asynchronous replication to maintain local copies of the entire data set. In a case of a network partition between the sites each site can continue to read/write its own data even when other sites are not available. It can also read other sites data but will not be able to change any data that is owned by other sites. In that context we give away some level of partition tolerance for consistency.
  • Availability: Local failures are dealt with through the same model as described above. Updates on data that belong to other sites will be blocked. In that context we give away some degree of availability for consistency.

Note – Another option to deal with consistency under concurrent updates between two sites can be based on versioning. In that case it is assumed that the latest version wins or the system will delegate the decision for resolution on conflicting updates to the application.  This is a more complex scenario that goes beyond the scope of this post.

Handling Extreme Failure:

Multiple node failures at the same time. In cases where both the primary and backup of a given partition fails at the same time, we will consider this cluster as non functional and block operations to that cluster until the cluster becomes functional again. In that case we will compromise on certain aspects of Availability for the sake of Consistency. If your application can tolerate the potential consistency issues associated with such failure, you could configure the cluster to route all  operations to currently available partitions. In that case we will trade Availability over Consistency – this behavior can be useful in streaming scenarios where the system is used to pass through events.

Note that if we choose to use a proper disaster recovery site setup where we ensure that primaries and their backups never run on the same site the chances for having such a failure is close to zero in the first place and can be considered equal for having the two sites down at the same time.

Recovery from total failure. When the entire system crashes, it will boot itself from the long-term persistent storage or from a snapshot (new). In that case we may lose some of the data that was kept in memory and wasn’t yet delivered to the  long term persistent storage. It is important to note that in a disaster recovery setup that event can happen only when both sites goes down at the same time.  In that case the system would recover itself from the data that was last updated in the long-term storage.

Final words

In many of the recent discussions on the design of large scale systems (a.k.a. Web Scale) it was argued that the right set of tradeoffs for building large scale systems would be to give away Consistency for Availability and Partition tolerance. Those arguments relied on the foundation of the CAP theorem developed in early 2000-2002. One of the core principals behind the CAP theorem is that you must choose two out of the three CAP properties. In many of the transactional systems giving away consistency is either impossible or yields a huge complexity in the design of those systems. In this series of posts, I've tried to suggest a different set of tradeoffs in which we could achieve scalability without compromising on consistency. I also argued that rather than choosing only two out of the three CAP properties we could choose various degrees of all three. The degrees would be determined by the most likely availability and partition tolerance scenarios in our specific application.  The suggested model was based on the experience we had in GigaSpaces over the course of the past years and was successfully deployed in many mission critical systems today in Finance, Telco and ecommerce business. I hope that through the sharing of this experience we could come up with a broader set of patterns on how to build large scale systems that would fit also to mission critical transactional systems.

References

• NoCAP – Part I

• NoCAP – Part II Availability and Partition tolerance

• Elasticity for the Enterprise -- Ensuring Continuous High Availability in a Disaster Failure Scenario

Scale-out vs Scale-up

In my previous post Concurrency 101 I touched on some of the key terms that often comes up when dealing with multi-core concurrency.

In this post I'll cover the difference between multi-core concurrency that is often referred to as Scale-Up and distributed computing that is often referred to as Scale-Out model.

The Difference Between Scale-Up and Scale-Out

One of the common ways to best utilize multi-core architecture in a context of a single application is through concurrent programming. Concurrent programming on multi-core machines (scale-up) is often done through multi-threading and in-process message passing also known as the Actor model.Distributed programming does something similar by distributing jobs across machines over the network. There are different patterns associated with this model such as Master/Worker, Tuple Spaces, BlackBoard, and MapReduce. This type of pattern is often referred to as scale-out (distributed).

Conceptually, the two models are almost identical as in both cases we break a sequential piece of logic into smaller pieces that can be executed in parallel. Practically, however, the two models are fairly different from an implementation and performance perspective. The root of the difference is the existence (or lack) of a shared address space. In a multi-threaded scenario you can assume the existence of a shared address space, and therefore data sharing and message passing can be done simply by passing a reference. In distributed computing, the lack of a shared address space makes this type of operation significantly more complex. Once you cross the boundaries of a single process you need to deal with partial failure and consistency. Also, the fact that you can’t simply pass an object by reference makes the process of sharing, passing or updating data significantly more costly (compared with in-process reference passing), as you have to deal with passing of copies of the data which involves additional network and serialization and de-serialization overhead.

Choosing Between Scale-Up and Scale-Out

The most obvious reason for choosing between the scale-up and scale-out approaches is scalability/performance. Scale-out allows you to combine the power of multiple machines into a virtual single machine with the combined power of all of them together. So in principle, you are not limited to the capacity of a single unit. In a scale-up scenario, however, you have a hard limit -– the scale of the hardware on which you are currently running. Clearly, then, one factor in choosing between scaling out or up is whether or not you have enough resources within a single machine to meet your scalability requirements.

Reasons for Choosing Scale-Out Even If a Single Machine Meets Your Scaling/Performance Requirements

Today, with the availability of large multi-core and large memory systems, there are more cases where you might have a single machine that can cover your scalability and performance goals. And yet, there are several other factors to consider when choosing between the two options:

1. Continuous Availability/Redundancy: You should assume that failure is inevitable, and therefore having one big system is going to lead to a single point of failure. In addition, the recovery process is going to be fairly long which could lead to a extended down-time.

2. Cost/Performance Flexibility: As hardware costs and capacity tend to vary quickly over time, you want to have the flexibility to choose the optimal configuration setup at any given time or opportunity to optimize cost/performance. If your system is designed for scale-up only, then you are pretty much locked into a certain minimum price driven by the hardware that you are using. This could be even more relevant if you are an ISV or SaaS provider, where the cost margin of your application is critical to your business. In a competitive situation, the lack of flexibility could actually kill your business.

3. Continuous Upgrades: Building an application as one one big unit is going to make it harder or even impossible to add or change pieces of code individually, without bringing the entire system down. In these cases it is probably better to decouple your application into concrete sets of services that can be maintained independently.

4. Geographical Distribution: There are cases where an application needs to be spread across data centers or geographical location to handle disaster recovery scenarios or to reduce geographical latency. In these cases you are forced to distribute your application and the option of putting it in a single box doesn’t exist.

Can We Really Choose Between Scale-Up and Scale-Out?

Choosing between scale-out/up based on the criteria that I outlined above sound pretty straightforward, right? If our machine is not big enough we need to couple a few machines together to get what we're looking for, and we're done. The thing is, that with the speed in which network, CPU power and memory advance, the answer to the question of what we require at a given time could be very different than the answer a month later. 

To make things even more complex, the gain between scale-up and scale-out is not linear. In other words, when we switch between scale-up and scale-out we're going to see a significant drop in what a single unit can do, as all of a sudden we have to deal with network overhead, transactions, and replication into operations that were previously done just by passing object references. In addition,we will probably be forced to rewrite our entire application, as the programming model is going to shift quite dramatically between the two models. All this makes it fairly difficult to answer the question of which model is best for us.

Beyond a few obvious cases, choosing between the two options is fairly hard, and maybe even almost impossible.

Which brings me to the next point: What if the process of moving between scale-up and scale-out were seamless -- not involving any changes to our code?

I often use storage as an example of this. In storage, when we switch between a single local disk to a distributed storage system, we don’t need to rewrite our application. Why can’t we make the same seamless transition for other layers of our application?

Designing for Seamless Scale-Up/Scale-Out

To get to a point of seamless transition between the two models, there are several design principles that are common to both the scale-out and scale-up approaches.

Parallelize Your Application

1. Decouple: Design your application as a decoupled set of services. “All problems in computer science can be solved by another level of indirection" is a famous quote attributed to Butler Lampson. In this specific context: if your code sets have loose ties to one another, the code is easier to move, and you can add more resources when needed without breaking those ties. In our case, designing an application from a set of services that doesn’t assume the locality of other services is used to enable us to handle a scale-up scenario by routing requests to the most available instance.

2. Partition: To parallelize an application, it is often not enough to spawn multiple threads, because at some point they are going to hit a shared contention. To parallelize a stateful application we need to find a way to partition our application and data model so that our parallel units share-nothing with one another. 

Enabling Seamless Transitions Between Remote and Local Services

First, I'd like to clarify that the pattern I outlined in this section is intended to enable seamless transition between distributed and local service. It is not intended to make the performance overhead between the two models go away.

The core principle is to decouple our services from things that assume locality of either services or data. Thus, we can switch between local and remote services without breaking the ties between them. The decoupling should happen in the following areas: 

1. Decouple the communication: When a service invokes an operation on another service we can determine whether that other service is local or remote. The communication layer can be smart enough to go through more efficient communication if the service happens to be local or go through the network if the service is remote. The important thing is that our application code is not going to be changed as a result.

2. Decouple the data access: Similarly, we need to abstract our data access to our data service. A simple abstraction would be a distributed hash table, where we could use the same code to point to a local in-memory hash-table or to a distributed version of that update. A more sophisticated version would be to point to an SQL data store where we could have the same SQL interface to point to an in-memory data store or to a distributed data-store.

Packaging Our Services for Best Performance and Scalability

Having an abstraction layer for our services and data brings us to the point where we could use the same code whether our data happens to be local or distributed. Through decoupling, the decision about where our services should live becomes more of a deployment question, and can be changed over time without changing our code.

In the two extreme scenarios, this means that we could use the same code to do only scale-up by having all the data and services collocated, or scale-out by distributing them over the network.

In most cases, it wouldn't make sense to go to either of the extreme scenarios, but rather to combine the two. The question then becomes at what point should we package our services to run locally and at what point should we start to distribute them to achieve the scale-out model.

To illustrate, let’s consider a simple order processing scenario where we need to go through the following steps for the transaction flow:

1. Send the transaction

2. Validate and enrich the transaction data

3. Execute it

4. Propagate the result

Each transaction process belongs to a specific user. Transactions of two separate users are assumed to share nothing between them (beyond reference data which is a different topic).

In this case, the right way to assemble the application in order to achieve the optimal scale-out and scale-up ratio would be to have all the services that are needed for steps 1-4 collocated, and therefore set up for scale-up. We would scale-out simply by adding more of these units and splitting both the data and transactions between them based on user IDs. We often refer to this unit-of-scale as a processing unit.

To sum up, choosing the optimal packaging requires:

1. Packaging our services into bundles based on their runtime dependencies to reduce network chattiness and number of moving parts.

2. Scaling-out by spreading our application bundles across the set of available machines.

3. Scaling-up by running multiple threads in each bundle.

The entire pattern outlined in this post is also referred to as Space Based Architecture. A code example illustrating this model is available here.

Final Words

Today, with the availability of large multi-core machines at significantly lower price, the question of scale-up vs. scale-out becomes more common than in earlier years.

There are more cases in which we could now package our application in a single box to meet our performance and scalability goals.

A good analogy that I have found useful to understanding where the industry is going with this trend is to compare disk drives with storage virtualization. Disk drives are a good analogy to the scale-up approach, and storage virtualization is a good analogy to the scale-out approach. Similar to the advance in multi-core technology today, disk capacity has increased significantly in recent years. Today, we have xTB data capacity on a single disk.

PC hard disk capacity (in GB).The plot is logarithmic, 
 so the fitted line corresponds to exponential growth

Interestingly enough, the increase in capacity of local disks didn’t replaced the demand for storage, quite the contrary. A possible explanation is that while single-disk capacity doubled every year, the demand for more data grew at a much higher rate as indicated in the following IDC report:

Market research firm IDC projects a 61.7% compound annual growth rate (CAGR) for unstructured data in traditional data centers from 2008 to 2012 vs. a CAGR of 21.8% for transactional data.

Another explanation to that is that storage provides functions such as redundancy, flexibility and sharing/collaboration. Properties that a single disk drive cannot address regardless of its capacity.

The advances with the new multi-core machines will follow similar trends, as there is often a direct correlation between the advance in the capacity of data and the demand for more compute power to manage it, as indicated here:

The current rate of increase in hard drive capacity is roughly similar to the rate of increase in transistor count.

The increased hardware capacity will enable us to manage more data in a shorter amount of time. In addition, the demand for more reliability through redundancy, as well as the need for better utilization through the sharing of resources driven by SaaS/Cloud environments, will force us even more than before towards scale-out and distributed architecture.

So, in essence, what we can expect to see is an evolution where the predominant architecture will be scale-out, but the resources in that architecture will get bigger and bigger, thus making it simpler to manage more data without increasing the complexity of managing it. To maximize the utilization of these bigger resources, we will have to combine a scale-up approach as well.

Which brings me to my final point -– we can’t think of scale-out and scale-up as two distinct approaches that contradict one another, but rather must view them as two complementing paradigms.

The challenge is to make the combination of scale-up/out native to the way we develop and build applications. The Space Based Architecture pattern that I outlined here should serve as an example on how to achieve this goal. 

 

References

• Parallel processing patterns:

• Actor model

• Master/Worker

• Tuple Spaces

• BlackBoard 

• MapReduce

• Space Based Architecture.

• Space Based Architecture (GigaSpaces Code example)

• Managing Tera Scale with Cisco UCS

• NAS systems evolve to cope with unstructured data growth

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)

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:

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

• Architecture Summit 2008 - Write up
• Are Cloud Based Memory Architectures the Next Big Thing? 
• Scaling Out MySQL
• Actor Model and Data Grids
• Lessons from Pat Helland: Life Beyond Distributed Transactions
• Xtreme Transaction Processing under SBA terminology

• Social Media Kills the Database

• Amazon SimpleDB is not a DataBase!