Nati Shalom's Blog

Big Data In the Cloud Using Cloudify

Edd Dumbill wrote an interesting article on O’Reilly Radar covering the current solutions for running Big Data in the Cloud

Big data and cloud technology go hand-in-hand. Big data needs clusters of servers for processing, which clouds can readily provide.

Edd touched briefly on the role of PaaS for delivering Big Data applications in the cloud

Beyond IaaS, several cloud services provide application layer support for big data work. Sometimes referred to as managed solutions, or platform as a service (PaaS), these services remove the need to configure or scale things such as databases or MapReduce, reducing your workload and maintenance burden. Additionally, PaaS providers can realize great efficiencies by hosting at the application level, and pass those savings on to the customer.

Even though Edd’s article covers all the different forms of running Big Data on private and public clouds, the article focuses mainly on the public cloud offering from Amazon, Microsoft and Google.

In this post, I wanted to cover more specifically how I see the evolution of cloud application platforms (PaaS) to support Big Data. I’ll refer specifically to Cloudify which was designed primarily to support Big Data applications. 

 

 

Big Data in the cloud using Cloudify

Background

Most of the PaaS solutions out there started by focusing on simple web application deployments on Ruby, Java and Node.js. Unlike other PaaS solutions, when we designed Cloudify we picked Big Data as the primary target for Cloudify, and started by supporting popular NoSQL clusters such as Cassandra and MongoDB, as well as providing the equivalent of Amazon RDS by providing recipes for MySQL. Our goal was to make Big Data deployments a first class citizen within Cloudify. To this end,when you download Cloudify you'll notice that ALL of our examples comes with pre-integrated Big Data deployments.

There are couple of reason that brought us to make that decision: 

• Managing large data clusters is a core expertise at GigaSpaces

Most people know GigaSpaces for our In-Memory Data Grid solution known as XAP (eXtreme Application Platform). Over the past 10 years, as our customer deployments grew substantially, we realized that developing strong automation and cluster management is as critical as handling data consistency, performance, and latency in of our data-grid product. In a large cluster, if something breaks it’s going to literally be impossible to handle that failure through manual procedures.  For that reason, we developed lots of IP around automation of data cluster deployment which resulted in a unique self-managed data cluster.

• Cloudify is a natural evolution of GigaSpaces Data Cluster

When we built Cloudify it made a lot of sense to take the IP that we developed for managing GigaSpaces cluster and simply generalize it so it would fit with any other framework. In this way, we could leverage the years of experience  as well as development in this area, and gain a significant head-start.

• Big Data applications are complex

Big Data applications tend to be fairly complex, which makes them an ideal candidate for the sort of automation and management that Cloudify can offer.

• Big Data applications have a lot in common with XAP applications

Both need automation of data, failover and recovery, both fit into large cluster deployments, and both share similar partitioning and other clustering architecture. 

What makes Big Data platform different than any other application

Most of the existing orchestration systems were designed to handle stateless processes. Moving data is a completely different ballgame as you need to think of:

• Primary and Backup dependency 
• Availability - moving data without losing it.
• Moving processes to the data rather than the other way around.
• Data replication within and across sites 

Automating any of these processes through general orchestration tooling like Chef or Rightscale can become a fairly involved and complex process, with lots of pitfalls with handling edge scenarios for example the handshake process that is often involved when automating a data node failure, including a split-brain scenario.

In Cloudify we were able to curve out lots of that logic from the user, for example Cloudify will automatically ensure that primaries and backups won’t run on the same node or data center in case of disaster recovery. You don't need to do anything but tag your cluster nodes with a zone-tag. 

Managing Big Data applications != Managing Big Data storage

Managing data clusters is one thing. Being able to process the data is yet another challenge that we need to think about when we’re dealing with application platforms as I noted in one of my earlier post.

The main challenge is that quite often the management of the data processing logic is built on completely different scaling, availability and monitoring tools than the one used for managing our Big Data deployment. It turns out, that this silo thinking leads to a whole set of complexities starting from the inconsistency in having multiple managers, each determined in a different way when there is a failure or scaling event, and that quite often end up conflicting with one another. Having lots of moving parts is yet another challenge that makes the entire deployment pretty much a complete mess.

Built-in support for In Memory Stream based processing

In an earlier post, 5 Big Data predictions for 2012, Edd pointed out that streaming data processing is going to be one of the main trends for Big Data. 

Over the next few years, we'll see the adoption of scalable frameworks and platforms for handling streaming, or near real-time analysis and processing. In the same way that Hadoop has been borne out of large-scale web applications, these platforms will be driven by the needs of large-scale location-aware mobile, social and sensor use.

Being also part of XAP, Cloudify already comes with built-in support for streaming Big Data processing. This means that building your own Facebook or Twitter-like real-time analytics can be as a simple as writing only small scripts that handle the analytics counters. All the rest, i.e. scalability, availability, automation, cloud portability, management and monitoring, is covered by Cloudify as noted in this and this use case. 

Examples for Big Data applications running on Cloudify 

In the list below, I tried to put together a couple of references and examples that will make it easier for you to get started. The first reference points to a simpler scenario that will allow you to use Cloudify to deploy your Big Data database as a service. The other three references are full-application stack deployments that include the data-processing and web-tier of applications managed together with the Big Data database. 

Running a Big Data 'Database as a Service'

Cloudify comes with built-in recipes for Cassandra and MongoDB, as well as Solr (popular search engine), which makes it easy to deploy these database clusters on your local machine, data center or private/public cloud through a single command. In this way, you can use Cloudify to automate the database. 

Spring Travel application with Cassandra

Demonstrating a deployment of a JAVA-based commerce application with Cassandra as the database

The example includes recipes that provision the Cassandra database, create a schema, load the data, then spawn a Tomcat container, automatically injecting the reference to Cassandra, that then shows the custom management and monitoring of the application - all through a single command. A recent videocast showing how the travel application works on HP OpenStack cloud is available here.

Pet Clinic example with MongoDB

The Pet Clinic example does pretty much the same thing only using a sharded MongoDB cluster.

Twitter Real Time analytics example for Big Data 

The Twitter example  shows how you can attach real time stream-based processing for handling real live Twitter feeds, and how you can manage both the stream processing cluster and the Big Data cluster using Cloudify and run it on any cloud. The entire source code for this example is provided on Github.

Give it a try

To try out any of the examples you'll need to download Cloudify (latest) or (stable) build. Cloudify comes with all first three examples as part of the distribution under the recipes or examples directory. To try out these examples simply follow the steps from the Cloudify Quick Start Guide.  

 References:

• Big Data in the Cloud
• 5 Big Data predictions for 2012
• Big Data Application Platform
• Analytics for Big Data – Venturing with the Twitter Use Case
• Cloudify Download  (latest), (stable) 

Realtime Analytics for Big Data: A Facebook Case Study

via www.youtube.com

Java and Center Stage: Meet us at JavaOne 2011

Next week, I’m going to be in JavaOne. Unlike last year, where I expressed lots of skepticism about the way Oracle drives the Java Community, it seems that this year things are starting to get back on course – not that the skepticism has vanished completely, but at least there is a stronger sentiment that things are starting to settle down and take a more positive course.

It seems that now Java is taking center stage in the Cloud world, with more application platforms such as SalesForce/Heroku, VMware, Redhat/OpenShift, Cloud.com, JClouds, and obviously GigaSpaces providing a rich set of offerings that are based on Java.

One of the most interesting things in my opinion is that the combination of Cloud and Java brings new opportunities that until recently were unheard of:

 

• Heroku, which was very much Ruby-centric, added support for Java.
• GigaSpaces recently announced our partnership with Microsoft to bring Java to Microsoft's Azure community.

What’s interesting in those two examples is that the Cloud forces us to break away from the language wall. It brings communities that were living in completely isolated silos much closer together.

This forces a very different way of thinking about Java - and about the next generation Java platform specifically.

The new platform should be:

• Open for other languages (through integration points like Protobuff, Thrift, REST, and more, as well as the ability to run those languages)
• Embrace NoSQL implementations as an integral part of the platform
• Add support for dynamic data structures such as JSON, and schema-free document models
• Bring Dev and Ops closer by providing better exposure of operational aspect as part of the development frameowork
• Make cloud deployment a first class citizen

It seems that Java EE 7 is taking the right path in that direction. From Java EE 7 and Cloud Computing:

• The Java EE platform architecture is taking into account the operational impact of the cloud, more specifically by defining new roles, such as a PaaS administrator.
• The Java EE platform may also establish a set of constraints for PaaS-specific features, such as multi-tenancy, that deployments may have to obey. Applications may also be able to identify themselves as designed for cloud environments.
• All resource manager-related APIs, such as JPA, JDBC and JMS, will be updated to enable multi-tenancy. The programming model may be refined as well by introducing connectionless versions of the major APIs.
• Java EE will define a descriptor for application metadata to enable developers to describe certain characteristics of their applications that are essential for the purpose of running them in a PaaS environment. These may include being multitenancy-enabled, resources sharing, quality of service information, dependencies between applications.

What This Means

Java is regaining some of its luster from the last few years. It was being perceived as vaguely pedestrian and ordinary, and languages like Ruby and Python were being seen as more dynamic environments for rapid and scalable deployments.

Now we're seeing Java as the lingua franca of enterprise development. With so many environments being heterogeneous, a homogenous platform becomes a very desirable resource, and the JVM is able to run many languages in an integrated environment: Java, Ruby, Python, Scala, and Groovy, to name only a few, and many of these can run as compiled code and as scripts.

Even beyond the multiple programming language support of the JVM, you have very broad support for almost any number of remote procedure call mechanism you can think of. The old J2EE approach, where you defined your requirement and there was a specific way to fulfill it (which meant that every architecture was more or less nudged to the middle), is no longer a real limit.

If you can do it, you can do it in Java, without Java forcing you to sacrifice in the process.

This is a very powerful concept.

It puts your organization back under your control. This is desirable because you know your product and architecture better than a community process does. You know whether you need to support REST here, and SOAP there; you know you need to use NoSQL in this area and MySQL in that area.

A Remaining Challenge

Java EE 7 may be becoming aware of the new cloud-based environment, but it's still only one aspect of application deployment and design. What it does not address is the entire application environment.

This is where applications such as Cloudify come in. Cloudify treats every artifact involved with your deployment as a managed element.

Consider: when you deploy an enterprise application, you're not just deploying some operational code. That code has external requirements, like a database (or NoSQL warehouse); it might also rely on an Apache-based load balancer, or perhaps another independently-deployed artifact.

That's quite a burden on operations, because all of these are separately deployed and managed.

Cloudify, however, centralizes the management of each element your application is composed of, over the entire application's lifecycle, and provides a bridge to whatever environment you wish to use.

For example, you could have a MySQL database, used by a Java application hosted by a single Tomcat instance, fronted by an Apache httpd server, hosted on an internal Linux server. Cloudify can deploy and monitor each of those elements: the MySQL database, the schema and initial data; Tomcat; the Java application; Apache httpd, and the server upon which it all runs.

Now let's look at a larger deployment: a set of MySQL servers, a cluster of GigaSpaces XAP nodes, an instance of GlassFish hosting some web services, six .Net-based web services, and five httpd servers being load-balanced, deployed on a platform such as Microsoft Azure. From Cloudify's perspective, the deployment and management process is the same. You have more artifacts to deploy, and you're telling Cloudify to deploy on a different platform... that's it.

Cloudify can make sure each resource is running optimally, as well. If you can measure an attribute - such as CPU load, or disk space, or free memory - you can tell Cloudify to assert that some headroom exists, and what to do if a metric is exceeded.

For example, if you have a CPU running at 100% for an extended period of time, Cloudify can automatically deploy another instance of the component consuming the CPU, and configure load balancing so your application is more scalable, without human intervention.

Likewise, if you have four servers running a given process, all at 0% CPU load, you can reduce the number of containers with that component so that you're not overallocating resources, saving energy and money.

The key thought here:

You know how to solve your problem. We can help.

We'll be on the floor at JavaOne 2011, in booth #5002. We'll be happy to show you both XAP, our industry-leading application platform, and Cloudify, our deployment solution for the cloud-enabled world.

 

 

(This post was co-written with Joseph Ottinger, who unfortunately won't be able to make it to JavaOne this year. But he says that the best people in GigaSpaces will be there, so you're not missing out!)

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