Nati Shalom's Blog

Real Time Analytics for Big Data: An Alternative Approach

Lately, we've been talking to various clients about realtime analytics, and with convenient timing Todd Hoff wrote up how Facebook's realtime analytics system was designed and implemented (See previous review on that regard here).

They had some assumptions in design that centered around the reliability of in-memory systems and database neutrality that affected what they did: for memory, that transactional memory was unreliable, and for the database, that HBase was the only targeted data store.

What if those assumptions are changed? We can see reliable transactional memory in the field, as a requirement for any in-memory data grid, and certainly there are more databases than HBase; given database and platform neutrality, and reliable transactional memory, how could you build a realtime analytics system?

Joseph Ottinger and I discussed this, and this is what we came up with.

A Summary of History

To understand what a new design might look like, it’s often useful to consider a previous design. This is a very short summary of Facebook’s realtime analytics system.

First, it’s based on a system of key/value pairs, where the key might be a URL and the value is a counter. Thus, there’s a requirement for atomic, transactional updates to a very simple piece of data. The difficulties come from scale, not from the focus of the system.

The process flow is fairly simple:

• A user creates an event by performing some action on the website. This generates an AJAX request, sent to a service.
• Scribe is used to write the events into logs, stored on HDFS.
• PTail is used to consolidate the HDFS logs.
• Puma takes the consolidated logs from PTail and stores them into HBase in groupings that represent roughly 1.5 seconds’ worth of events.
• HBase serves as the long-term repository for analytics data.

There are some questions around how PTail and Puma serve as scaling agents, and some of the notes around their use are still limited in scale – for example, one of the concerns is that an in-memory hash table will fill up, which sounds like fairly serious limitation to have to keep in mind.

A Potential for Improvement

There are lots of areas in which you can see potential improvements, if the assumptions are changed. As a contrast to Facebook's working system:

• We can simplify the design. If memory can be seen as transactional - and it can - we can use them without transforming them as they proceed along our analytics workflow. This makes our design and implementation much simpler to implement and test, and performance improves as well.
• We can strengthen the design. With a polling semantic, such systems are brittle, relying on systems that pull data in order to generate realtime analytics data. We should be able to reduce the fragility of the system, even while making it faster.
• We can strengthen the implementation. With batching subsystems, there are limits shouldn’t exist. For example, one concern in Facebook's implementation is the use of an in-memory hash table that stores intermediate data; the in-memory aspect isn’t a concern until you realize that the batch sizes are chosen partially to make sure that this hash table doesn’t overflow available space.
• We can allow deployments to change databases based on their requirements. There's nothing wrong with HBase, but it's got specific characteristics that aren't appropriate for all enterprises. We can design a system which you’d be able to deploy on various and flexible platforms, and we can migrate the underlying long-term data store to a different database if needed.
• We can consolidate the analytics system so that management is easier and unified. While there are system management standards like SNMP that allow management events to be presented  in the same way no matter the source, having so many different pieces means that managing the system requires an encompassing understanding, which makes maintenance and scaling more difficult.

What we want to do, then, is create a general model for an application that can accomplish the same goals as Facebook’s realtime analytics system, while leveraging the capabilities that in-memory data grids offer where available, potentially offering improvement in the areas of scalability, manageability, latency, platform neutrality, and simplicity, all while increasing ease of data access.

That sounds like quite a tall order, but it’s doable.

The key is to remember that at heart, realtime analytics represent an events system. Facebook’s entire architecture is designed to funnel events through various channels, such that they can safely and sequentially manage event updates.

Therefore, they receive a massive set of events that “look like” marbles, which they line up in single file; they then sort the marbles by color, you might say, and for each color they create a bundle of sticks; the sticks are lit on fire, and when the heat goes up past a certain temperature, steam is generated, which turns a turbine.

It’s a real-life Rube Goldberg machine, which is admirable in that it works, but much of it is still unnecessary if the assumptions about memory ("unreliable") and database ("HBase is the only target that counts") are changed. Looking at the analogy from the previous paragraph, there’s no need to change a marble into anything. The marble is enough.

A Plan for Implementation

Our design for implementation is built around putting data and messaging together. A data grid is a perfect mechanism for this, as long as it provides some basic features: transactional operations, push and pull semantics, and data partitioning.

A data grid does provide those basic features, or else it's not really much of a data grid; it'd be more of a cache otherwise.

With a data grid, then, the events come in as individual messages. When the user chooses an operation on the web site, an asynchronous operation would write the event, just as Facebook does today. However, instead of filtering and batching the events into various forms, the events are dispatched to waiting processes that perform many transactional updates in parallel.

There’s a danger that those updates might be slower than the generated events, if each event is processed sequentially. That said, this isn’t as much a problem as one might think; if data partitioning is used, then event handlers can receive partitioned events, which localizes updates and speeds them up dramatically.

In fact, you can still use batching to process events as a group; since the events would be partitioned coming in, the batch process would still be updating local data very quickly, which would be faster than individual event processing, even while retaining simplicity.

With this design, there is no overflow condition, because a system that’s designed to scale in and out as most data grids are will repartition to maintain even usage. If a data grid can’t provide this feature intrinsically, of course some management will be necessary, but finding data grids with this feature isn’t very difficult.

One other advantage of data grids is in write-through support. With write-through, updates to the data grid are written asynchronously to a backend data store – which could be HBase (as used by Facebook), Cassandra, a relational database such as MySQL, or any other data medium you choose for long-term storage, should you need that.

The memory system and the database - the external data store - work together. The in-memory solution is ideal for the realtime aspects, the events that affect now. The external data storage solution is designed to handle long-term data, for which speed is not as much of an issue.

A Discussion of Strengths

The key concept here is that event handling is the lever that can move the realtime analytics mountain. By providing a simple, scalable publisher/subscriber model, you simplify design; by using a platform that supports data partitioning, transactional updates, and write through capabilities, you gain scalability.

The data grid’s flexible query API means that events can literally react when data is available.

For a call center, for example, you want to immediately identify signals that show that the caller should be handled differently; imagine an ecommerce site that was able to determine immediately if a user was losing interest, and thus could respond appropriately, before the customer moves on.

With external processes and a long funnel for data, immediate-response capabilities are very difficult to implement, not just because of latency but because the data transformations tend to homogenize the data, instead of allowing rich expressions and flexible event types.

The data grid also has much richer support in terms of client applications. Instead of applications going through an API that focuses on a specific phase of the data’s life (for example, an API focused on HBase), you can focus on a generic API that can capture events at any point in their lifecycle, and from anywhere. An external monitoring process, then, can have the same immediate, partition-aware access to data that the integrated message-handling system does; adding features and analysis is just a matter of connecting a client to the data grid.

Here we have a quick demo that shows much of this in motion. We have a market analysis application, deployed into GigaSpaces XAP via our new Cloud deployment too, Cloudify; it uses an event-driven system to display realtime data, with a write-through to Cassandra on the back-end. The design is very simple, and demonstrates the principles we've discussed here - and can scale up and down depending on demand.

Final words

As real-time-analytics gets into mainstream application design, it becomes important to draw a blueprint on how to build this sort of system without re-inventing the wheel.

Todd Hoff (HighScalability) and Alex Himel (Facebook) provided a fairly detailed description on their solution and even more importantly they even shared the rationales that made them do things in certain ways.

One main difference in assumptions that lead to the different implementation strategies are in reliable memory for event processing, and in the use of passive data storage.

Another difference is that we had to to think of the solution as an easily cloneable solution and therefore a lot of attention was put on the simplicity of the runtime, packaging and management of the solution.

Yet another difference is that we couldn’t decide on a specific database as there isn’t a "one size fits all" solution – for certain customers, SQL would still be preferred choice and the fact that we can buffer the write to the database gives them more headroom while still allowing them to scale on writes.

I hope that this would lead to constructive dialogue on the various tradeoffs which will serve the entire industry...

References

• Video: Realtime Analytics for Big Data: A Facebook Case Study

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)