Posted tagged ‘Scalability’

Remote Datasources

February 18, 2010

Almost all applications today deal with persisted data. The only rare exceptions are usually not very complicated such as a calculator, unit conversion tool, or minor utility. Let us assume that any application we’re working on is going to have some sort of persistant state that must be kept between executions of the application. Typically the state is stored in either a file or a database. File access is useful for many desktop applications used by a single user as the format can be created very easily from the data model, the data is easy to transfer to another user via email or some other file transfer process, and backups involve just storing the file somewhere safe.

Many applications that involve multiple users, including web applications, will instead employ a database. Even single user desktop applications sometimes use a database. Database access is often performed by a section of the code converting objects or structures into database tupils and storing them in tables. When the data is required again, select statements are made that call the data and turn it back into a structure. Usually this storage and retrieval of data is done in the same code base as the rest of the application.

There are a few problems with this approach however. For one, the database code can sometimes become tightly coupled with the business logic of the application itself. The application knows about the data being stored in a database and will often be changed accordingly to facilitate database access. Switching to a different database system can involve very large rewrites of not only the database code, but also application code as it can rely on database specific functionality. Think of functions that pass along snippets of SQL in order to construct the correct objects. Often that SQL will be database specific.

Besides being tied to a single database, the same coupling also ties applications to databases in general. If your application is expecting the data to be stored in a relational database, assumptions in the business logic will be made regarding that. This inhibits your ability to transition to a file store or other NoSQL solution if required. Because of these concerns, my topic for today is about abstraction and remote data sources.

If you code your database logic in with your application code, coupling can form. However if you separate them completely your application doesn’t need to know where the data comes from or where it goes when its saved. All the application has to do is follow a simple generic interface for storing and retrieving data. This not only elements coupling, but has some other scalability benefits that I’ll discuss a bit later.

To really appreciate interfaced based design and loose coupling, try removing your data access code from your application code base completely. Instead, use a language neutral format such as JSON or XML to communicate between your application and your data source, itself a small application. The application will send a request along with perhaps some data if it needs something stored. The data source will then transform that data into a format that can be stored and place it either in a database, file, or any other storage medium that you desire. Later if you want to modify how your data is stored, maybe a different DB system is more suited or you’re moving from a file-based system to a database, you can just modify the implementation of the data source application while not changing a line of code in your actual application. The loose coupling means that as long as the data source maintains a solid interface that the application can interact with, the application can be developed independently of the data source.

If each of your interactions between your application and datasource are atomic and independent of each other, stateless, you can horizontally scale your datasource using replication. Each data source can access a different replicated read-only data store for request information. Meanwhile write access will all go to a master data store that can be replicated to the read-only slaves. Read requests from an application can be distributed to multiple data sources, especially useful if the data source does any CPU intensive filtering or processing.

The main benefit however remains that by separating your data source from your application as a remote service, you can ensure that your code is loosely coupled and that your business logic will not rely upon the underlying method which your data is stored. This will lead to greater flexibility in your code, and a more agile platform on which to build.


Scalability Testing with Virtual Machines

February 8, 2010

Making a service based, distributed application is a good way to help your scalability goals. If you can compartmentalize the different aspects of your application into separate processes that can host services through either web services or plain sockets, you’ll gain the ability to make asynchronous and parallel calls between services.

When you’re developing an application you generally use a single development machine. You can run into problems when trying to do integration tests between multiple services due to the limits placed on listening to ports and how some technology stacks can act differently on the same machine in comparison to using multiple machines. While your application might be distributed onto many servers, your development environment might not have the same number of servers to access, often due to budget constraints. This is especially true if you’re just hacking something up in your spare time at home or if you’re a self-employed programmer. However, machine resources can be limited in many companies due to the cost of those servers.

Good news is you probably don’t need additional physical servers just for your own testing. Most computers today are so powerful that you don’t use all of the resources of it very often. You can harness those additional resources by running virtual machines and using those as your additional servers. A really good place to acquire a high-quality, free virtual machine package is from Virtual Box. It’s an open source virtual machine started by Sun and now owned by Oracle that’s quite fast, multiplatform, and of course free.

The main idea will be to run multiple virtual machines, each one representing a proxy of a real server you would run your application on. You can then distribute your services onto these VMs and test how your interactions work. Each VM will have less resources than a real machine, but still enough for most testing you’ll need to do. In fact you can use VMs on your real servers if you find your resource utilization on those servers are low, but you need more machines for your application design.

This is also useful if you’re attempting to build a fault-tolerant distributed application where a single service might be replicated onto many machines. Any distributed application needs to deal with node failure, which is difficult to simulate on a single box. However, if you run your service on multiple virtual machines, you can see the effect of shutting one down during execution to see if your clients can handle the service outage and redirect to a running service. This can also be used to test the reintegration of a previously down server back into a distributed service pool.

Of course this isn’t new stuff at all. Many people have been using VMs for years for a variety of purposes. I’d like to mention however that if you’ve avoided VMs because you don’t see how they can apply to your work, if you think your work is too small for it, you should reconsider. Even a non-developer can benefit from using Virtual Machines. Sometimes you might want to run a piece of software that doesn’t work in your current OS, but you’d like to gain its services. You can fire up a VM of the exact OS that software supports and run it inside of a virtual machine. Some VMs can be fairly light-weight, taking up only a minor amount of ram, disk space, and cpu usage.

Another developer usage for VMs is to make a completely clean build environment. If you want to make sure your package can build, run, and execute with many Linux distributions you can create automations to create new virtual machines, install the OS, import your code, compile, test, run your application, export the results, then remove the virtual  machine. You can use VMs as part of an extended systems test for your application. This way you can find out if your application relies on some odd configuration that might cause difficulties for your users. It also can help you refine your build process and configurations for multiple distributions or Operating Systems.

In the current development climate, processors have nearly reached their speed limits. It is said that the next big push is for more parallel processing among multiple cores. Beyond that is distributed computing. When needs reach beyond the multiple processors of a single machine, we’ll look at the resources of multiple processes of multiple machines. This work is already being done at Yahoo!, Google, Microsoft, Amazon, Ebay, Facebook, and all of those other large tech corporations. However in the future I suspect that it will become fairly common to utilize the resources of many devices for a single application. Virtual Machines can give you a cheap way to explore the exciting realm of distributed applications from the comfort of your single (hopefully mutli-core, multi-processor, or both) machine.

Client Side Map-Reduce with Javascript

January 29, 2010

I’ve been doing a lot of scalability research lately. The High Scalability website has been fairly valuable to this end. I’ve been thinking of alternate approaches to my application designs, mostly based on services. There was an interesting article about’s architecture that describes a little bit on how they put services together.

I started thinking about an application that I work on and how it would work if every section of the application was talking to each other through a web service or sockets passing JSON or Protocol Buffers rather than the current monolithic design that uses object method calls. I then had the thought that why limit your services to being deployed on a set of static machines. There’s only so much expandability in that, what if we harnessed all of the unused power of the client machines that visit the site.

Anyone who’s done any serious work with ECMAScript (aka Javascript) knows that you can do some pretty powerful things in that language. One of the more interesting features about it is the ability to evaluate plain text in JSON format into Javascript objects using eval(). Now eval() is dangerous because it will run any text given to it as if it were Javascript. However that also makes it powerful.

Imagine needing to do a fairly intensive series of computation on a large data set and you don’t have a cluster to help you. You could have clients connect to a web site that you’ve set up and have their browser download data in JSON format along with executable Javascript code. The basic Javascript library can be such that a work unit will be very generic and contain any set of data and functions to perform on that data along with a finishing AJAX call to upload the results. On the server side when you have a Map-Reduce operation that you need to perform, you can distribute work units that contain both a section of the data along with the code needed to execute on it to any connected clients that have this library and are sending AJAX polling requests asking for work.

A work unit gets placed into a JSON string and sent back as the AJAX reply. The client then evaluates the reply which calls a javascript function that processes the data (which is probably a map job). Once the data is process the javascript automatically makes another AJAX call to send the result data back, most likely with some sort of work unit ID to keep track of anything, and requests the next work unit. The server then just coordinates between the various clients, setting time outs for each work unit so that if someone closes their browser the job can be handed out to an active client or a back-end server.

This will work a lot better on CPU intensive processes than it will on memory intensive ones. For example, sorting a large data set requires some CPU, but a lot more memory because you need to keep each element in memory that you’re sorting. Sending entire large lists to clients to sort and receiving the results would be slow due to bandwidth and latency restraints. However, performing large computations on a smaller series of data such as what’s done with SETI or brute force cryptography circumvention where you can send heartbeats of partial results back, there could be a benefit.

The limits of course will be on how much memory you can allocate in your browser to JavaScript. Also, since this technique would focus on heavy computational functions, the user will probably notice a fairly large hit on browsing speed in non-multithreaded or multiprocessing browsers. Naturally from a non-scientific point of view, most people would be outraged if their computer was being taken advantage of for its computing resources without their knowledge. However for applications working on an internal network or applications that state their intentions up front, users might be interested in leaving a browser open to a grid service to add their power to the system. This would probably be easier if you make each work unit worth a set of points and give them a running score like a game.

People like games.

Breaking a monolithic applicaiton into Services

January 22, 2010

Today I shall be tackling Service Oriented Architecture. I think that that particular buzz phrase has annoyed me a lot in the past. CTOs and middle management talk about how the read in a magazine something about SoA or SaaS (Software As A Service) as being the next big thing and that we should switch all of our stuff to that! Meanwhile there isn’t a very good explanation of how or why to do such a thing. I had written it off mostly as I never saw how independent services could be pulled together into an application. Sure it sounded neat for some other project that wasn’t mine. But no way did that model fit in with the project I’m doing.

Recently however I’ve been tasked with rewriting the section of code that generates reports in our application. The entire application is just a series of Java classes nested together and packaged up as one WAR file that is deployed to a Tomcat server. To scale, we deploy the same war file to another server (ideally everything is stateless and share-nothing so there’s no sessions to worry about). The whole thing is behind a load balancer that uses a round-robin strategy to send requests to various machines. Seems like a pretty good way to scale horizontally I thought.

However the horizontal scaling of this system is fairly coarse grain. If we find that getting hit with a lot of requests to generate reports is slowing down a server, our option is to duplicate the entire application on another server to distribute the load. However that is distributing not only report generation but also every other aspect of the application. So now we have an additional front end, service, and database layer running just to speed up one area. It seems like a bit of a waste of resources.

So instead of scaling a monolithic application, how about we break the whole thing up into ‘services’. Instead of the Report system just being a set of APIs and implementations in the applications source code, we instead make an entire new application that just generates reports as its whole goal. It has a basic interface and API that takes in a request with certain criteria of what type of report you want and in what format and returns that report. It can be a completely stand-alone application that just sits and waits, listening on a port for a request to come in. When one does it processes the request, generates a report, then sends the information back over that port to the client and waits for the next request. Naturally we make it multi-threaded so that we can handle multiple requests at a time, putting a limit on that number and queuing any overflow.

The benefit of this is manyfold. For starters you gain fine-grain horizontal scalability. If you find that one service is too slow or receives a lot of requests you can just deploy that service on additional machines. You only deploy that service however rather than the whole application. The service can be done via RPC, direct sockets, web services, or whatever else you like to listen for your requests. A controller near the front end would just call each individual service to gather the end data up to pass back to the user. Put it behind some sort of load balancer with failover and you have fine-grain horizontal scalability.

Second, you gain faster deployment. Since each service is self-contained they can be deployed independent of the other services. Launching a new version does not have to replace the entire system, just the services that you’re upgrading. In addition, since each service can be running (and should be running) on multiple machines, you can upgrade them piecemeal and perform bucket tests. For instance, if you have a service that is running on 10 machines you can upgrade only 2 of them and monitor user interaction with those services. This way you can have 20% of your users actually utilize your new code, the rest will hit the old code. When you see that everything is going fine you can upgrade the other 8 servers to have the new version of the service.

Because each part of the program is just a small separate program itself, problems become easier to manage and bugs become easier to fix. You can write tests for just the service you’re working on and make sure that it meets its contract obligation. It helps manage the problem with cyclical dependencies and tightly coupled code. It also reinforces the object-oriented strategy of message passing between components.

Further more, try to avoid language specific messages such as Java Serialization for communicating between your services. Utilize a language agnostic format such as JSON or Google’s Protocol Buffers. I would avoid XML because it’s fairly verbose and slower to transmit over a network and parse than those others. The advantage of using a language agnostic format is that each service can be written in a different language depending on the needs of the service. If you have a service that really needs to churn through a lot of data and you need every ounce of speed out of it you can write it in C or even assembly. For those areas where it would be good to do something concurrently you might use Scala, Erlang, or Clojure. The main idea is each part of your program can be written in a language that is best for solving that services problem.

Two Databases, One Hibernate

January 13, 2010

Hibernate is a very interesting ORM product that hides the SQL behind mapping your objects to your database. In general it works pretty well at making it easier to deal with database objects. Unfortunately, there seems to be some lack of flexibility. For instance, utilizing two databases for the same dataset in Hibernate.

Here’s my issue: I have a MySQL database that acts as a primary master. There is a slave database that is replicated off of this first database (to make it easier to talk about, I’ll call the master db “db1” and the slave “db2”). Db2 has an up-to-date copy of db1 and is set in read-only mode because we only ever want changes to happen in one location (db1). However, due to database load or scaling we want to modify our application to send all read-only requests to db1 while sending write requests to db2. Since Hibernate abstracts out the database fairly well, I’m not sure how to accomplish this.

In Hibernate you can read an object from the database, make modifications to that object, and Hibernate will seamlessly save those modifications right back into the database. How then do you force Hibernate to read objects from one DB but save them to another? You obviously can’t use the same session, which hibernate is centered around. However attaching a persistent object to a new session of a different database doesn’t seem very feasible either. I suppose there are perhaps some sort of interface or abstract class I can implement or extend to plug in, but there’s no documentation that I’ve found that deals with this issue.

The hassle of losing this flexibility makes me wonder if dealing with an ORM suite is really a benefit. Portability seems to decrease (as much as database agnostic Hibernate’s queries are) because I’m still tied to the one database model. The feature list of the framework talks about “extreme scalability” and that it may be used in a cluster, yet the documentation does not make any mention of this supposed feature. It almost seems like a bullet point that someone placed on the hibernate website as a marketing point that they later expected no one to really look into. Perhaps it can be adopted into a cluster yet out of the ‘box’ functionality seems to be missing or hidden.

I’m not really sure if ‘cluster support’ would really be of any benefit to us anyway. We’re not running what would be called a ‘cluster’. We’re using multiple front-end machines behind a load balancer that handle stateless requests. There is no coupled connection between the application servers. None of them know about any other application server running as there is no clustering software that would do something like share session data. Having a decoupled stateless database system where we can separate reads and writes to different (though content-wise identical) databases does not seem to fit into Hibernate’s current scope.

If anyone has any insight onto this matter I’d really appreciate it. I realize that there’s many aspects of the framework that I have not delved too deeply into and it is very possible that there are interfaces and/or abstract classes that I can implement and/or extend to gain the required functionality I desire. However the pickings will be slow due to lack of decent documentation in this area. I shall instead have to resign myself to trolling internet forums about Hibernate usage in Java, looking for nuggets of useful truth among the squalid remains of ORM flamewars.


January 4, 2010

I’m not talking about the type that is done at the airport, but more the type that lets you see how long or how many times each section of code is called. There is most likely a wide variety of tools for profiling your code in a variety of languages. I’m going to focus on Java in this post using a tool called YourKit, but I assume other tools will give the same basic info.

From time to time, especially when you’re having performance problems, it’s a good idea to run your code through a profiler and gain a readout of how long methods take to run and how many times they are called. A good profiler will break it down into a nice tree view that shows which methods took the longest and which methods inside of those methods took a long time based on the number of invocations. Looking at these numbers you can sometimes find places where your code has a nested loop that isn’t required or is calling an expensive method or function that returns the exact same data multiple times.

A developer is trained to trace code in his/her head while writing it, however sometimes you write a method that uses the functions of an already written piece of code. If you don’t know how efficient that code is, you may place it in a loop without realizing the performance penalty. When testing it locally it might seem to run fine because you’re a single user with very low load. A method that called the DB 10k times might return in seconds and seem to display fine for you, but under load it can slow down in production.

While a load test suite can find that a problem exists in the code, profiling can narrow it down to what’s really causing it. You can profile your code without a commercial or open source profiler by manually adding timestamp information before and after method calls. This isn’t very usable to do for every single call, but if you have an idea of what areas are causing a slow down and you want to narrow it down it can be very helpful.

Recently I ran a profiler on a section of a project I’m working on. I was able to find that to load a fairly simple page that there was a database query being called 22k times. That DB query was pulling in information related to an item I actually used, but was never used itself. While those queries return fairly quickly normally, having multiple users hit the page at the same time will multiply the number of queries. Also, if the latency between the server and the DB increases, we will definitely see a decrease in the responsiveness of the application. If we can eliminate these unneeded DB calls, we can speed up the application for the end users and also increase scalability.

Just because you don’t see an issue when you’re running your quick visual test doesn’t mean that the issue doesn’t exist. Performance is not something that should be tacked on at the last minute, and code needs to be analyzed regularly to see where inefficiencies are introduced.

Apache Pig Tips #1

December 2, 2009

Pig is a new and growing platform on top of Hadoop that makes writing jobs easier because you can avoid writing Map and Reduce functions in Java directly while still allowing you to do so if you choose. Instead it creates a bunch of basic functions such as COUNT, FILTER, FOREACH, and such that you would normally have to independently write for each data manipulation you want to perform. Unfortunately, the Pig documentation is fairly sparse and performing what you would think is a basic manipulation can become very difficult if there are no examples.

In this post, I’m going to provide some examples based on what I have learned about Pig in the last week. I labeled this as Apache Pig Tips #1 because I expect I may write more in the future as I uncover additional usages.

My problem domain includes a data set that has multiple IDs and a result field:

{tcid, tpid, tsid, date, result}

There are a few more fields but I’ll leave those out for brevity. A quick description of what each of those IDs are: the tcid is a Test Case id that a result was inserted for. The tpid is the Test Plan that the Test Case was a part of for this result. The tsid is the Test Suite that the Test Plan belongs to. The date is the date the result was added, and the result is the actual result (Pass, Fail, Postponed… etc).

Now a Test Plan can have multiple Test Cases in it, however it can only have each test case in it once. A Test Case can also be in multiple Test Plans (though again only once for each Plan). A Test Suite can have multiple Test Plans, but each Test Plan belongs to exactly one Test Suite. Results for a test case in a test plan can be inserted multiple times. Maybe the first time it was tested it failed so a Fail entry is added. At a later date it passes so a new entry is  made with a Pass result. We need to generate a report that shows how many Pass and Fail per test suite using only the latest result (ignoring previous ones).

The tab separated data is located on HDFS in a file named ‘allresults’. First we need to load the data into a variable:

A = LOAD 'allresults' USING PigStorage() AS (tcid:int, tpid:int, tsid:int, date:chararray, result:chararray);

Next we need to find all Test Case/Test Plan combinations and group by them. This will give us a list of items that has multiple results of different dates, but all for the same test case in a test plan.

B = GROUP A BY (tcid, tpid) PARALLEL 100;

The Pig documentation mentions that the GROUP keyword can be applied to a single alias and the BY can apply to a single item. What isn’t easily discovered in the documentation is that the item can be a tuple, which you can define in line by surrounding multiple fields with (). Normally your group by looks like: B = GROUP A BY tcid; However, to group on multiple fields so that each entry is a unique combination of those fields you can surrounded it with () to make it a tuple.

Currently the definition of B looks something like this:

{group {tpid:int, tcid:int}, A {tcid:int, tpid:int, tsid:int, date:chararray, result:chararray}};

Basically we have a Bag where each item in the bag has a Bag containing the unique tpid and tcid, along with a second bag that contains 1 or more result rows. We need to look at that second bag and remove all but the most recent result rows so that we have just the most recent result.

    X1 = ORDER A BY date;
    X2 = LIMIT X1 1;

This will loop through all items that were grouped by tcid and tpid. For each one it will order the inner result bag by date (descending by default). We then take only the first item from each of those ordered bags (the most recent result). We export to X the flattened version of the limited bag. This produces just a Bag of tuples that have all the non-recent results removed.

After that we can split up all of the results into separate aliases by filtering on X multiple times:

pass = FILTER X BY result == 'Pass';
fail = FILTER X BY result == 'Fail';
postpone = FILTER X BY result == 'Postpone';
-- group by suite and count it
passbysuite = GROUP pass BY tsid PARALLEL 100;
failbysuite = GROUP fail BY tsid PARALLEL 100;
postponebysuite = GROUP postpone BY tsid PARALLEL 100;
-- generate counts
passcount = FOREACH passbysuite GENERATE group, COUNT(pass);
failcount = FOREACH failbysuite GENERATE group, COUNT(fail);
postponecount = FOREACH postponebysuite GENERATE group, COUNT(postpone);

We now have 3 Bags, each containing a group which represents the tsid, along with a number correlating to how many results of each type existed.