Kopfhaus Software and Speed

Scaling Data, Part 1 – Scaling Relational Databases

This is my first post in a series of discussions on scaling applications. This applies to everything from web sites to corporate intranet software. Most of the context will be in terms of web applications, since that's what I've recently been working on. The first focus on this series is going to be data. Whether the data is generated by your employees or by your users, modern applications tend to have excessive amounts of it. Users expect to see this data in meaningful ways, as do managers, investors, and analysts. They want it all, and they want it now. So the question becomes, how do we handle all of this data as the application continues to grow? Beyond that, how fast can we make it? I'm going to do my best to give you meaningful answers to these questions.

These posts are geared toward developers, and cover a few issues in painful detail, but glance over a few very difficult topics that you'll have to research on your own. I'm going to continue to build on all of these examples and topics in later posts.

Data Relationships

Data gets it value through context. For instance, take the integer 12. By itself, it means next to nothing, unless you have numerophobia, but if I tell you that's how many purchases people have made in the last 10 minutes, or that's how many days overdue your credit card bill is, that number has a relative value beyond it's definition. These simple relationships are the main way we give data context, and database systems have been continually evolving the way we can express the relationships between data for half a century.

If you haven't read it yet, please take some time to look over The Pizza Question, specifically Solution 2.

The standard solution for these types of problem is a Relational Database Management System (RDBMS), something like MySQL, MsSQL, Oracle, PostgreSQL, etc. These systems all have their individual strengths and weaknesses, but are essentially based off of the same ideas of how data should be stored, organized, and accessed. They even share a (mostly) standardized method of accessing these features through the Structured Query Language (SQL).

Solution 2 is almost completely reliant upon the RDBMS to extrapolate meaning from the data. The RDBMS enforces constraints (foreign keys), concatenates data (joins), and handles every operation that we could possibly want to enact on our data. It's a pretty good deal for us, as engineers, to be able to take advantage of all of this management, as we don't have to write any of those pieces of code ever again. This is the tried and true approach that has been used for as long as I have ever known, and it works very well.

Relationship Issues

Solution 2 works in practice, the data seems flexible and the code wasn't too difficult to write, but there is a major issue. The issue is computational complexity, which is usually represented in Big-O notation. Every time we select the oldest unprocessed Order, the database has to search all Order records to find the first one that matches, this is dependent on the number of order records. If it takes .001 seconds to find a matching record from the Order table when there are 1,000 Order records, it could take up to 1 second to find each record at 1 million rows, or 10 seconds at 10 million rows. This is an Order-n complexity, O(n), where n is the number of records that need to be analyzed. This means that as the size of n grows, the time will also grow at a proportional rate. The idea of 10 million rows might seem preposterous for a pizza ordering system, but it would probably make sense if you were building a system for an international pizza conglomerate.

Now there are things much worse than an O(n) complexity, for instance O(n^2) or O(2^n), but there are also ways to much better than O(n), and it's these methods that have to take advantage of when n becomes too large to be handled by traditional means. For instance, a binary-tree index can reduce search time (on the indexed field) to O(log n), and a mapping algorithm can reduce searches to O(1) which is a constant time. These approaches are exactly what the underlying RDBMS attempts to do when you index a column.

The Big-O notation is not an exact measurement, but a rough estimation of the complexity of the operation. As anyone who's attempted to scale an application can tell you, the "trivial" factors like network overhead, process overhead, and context switching can add up so fast it will make your head spin. You must also consider that the database must do a join when you request information spread across multiple rows or tables. All of these things increase the complexity, and therefore increase the response times of the RDBMS, inching you ever closer to that point where things just stop working, or things get so slow that customers take their business elsewhere.

Continuing with the pizza conglomerate example, let's say that you receive 100k unique orders per day. I'm not going to focus on the "static" data, I'll go into caching in a later article, for now I'll focus on user-generated-data. During the order process, we'll estimate that the order information is accessed 3 times on average (1 insert, 1 select, 1 update).  That means the database system has to do about 3.4 transactions per second. This is perfectly reasonable with a single RDBMS server with proper configuration. But this doesn't give us the entire picture, for instance: Inserts are usually faster than updates, but they can cause locking issues. There are also external factors, and these are the ones that will cause the most trouble. The site you've built for the pizza conglomerate is doing so well, that the boss decides to buy some traffic on the most popular social network. Anyone talking about pizza, browsing pizza fan groups, or even thinking about pizza is going to see an advertisement for your site, all for a nominal fee. Overnight, the traffic doubles. The RDBMS begins to slow at first, but as the number of records increases, the load increases, and it eventually stops responding. Now we have to figure out why and how to alleviate it.

We know that RDBMS systems utilize RAM to store commonly accessed indexes and tables. Once the size of this data exceeds the size of the physical memory on the server, the RDBMS will need to utilize clever strategies to try and keep access times down, but at some point they will break down and the machine will need to use virtual memory, meaning disk swaps. The problem is that memory issues can cascade. As requests come in, the server still accepts them until it's input buffer fills up, but each request gets queued up. If a request takes too long to elicit a response from the server (say 15 second timeout), "smart" code will usually attempt to get the data once again, because the code needs that data to continue on. This means we now have multiple copies of every query in the queue as they're waiting to be answered. Once the input buffer fills up, the server will stop responding. Even once all of the connections are eliminated, after a warm-up period, the server will still go into a state of "thrashing" where it just can't keep up anymore, even if traffic is reduced.

So what do we do? First, we need to analyze how to reduce the number of accesses, and make sure that we actually need to access the data 3 times per order. This is a topic for caching and optimization, and I'll cover this later. Assuming data access is already optimized, we need more hardware, but how do we utilize it? The easiest possible solution is to move different tables to different servers. But wait, if you do that, most RDBMS systems won't let you do joins across multiple servers, and so far, we're reliant on the RDBMS enforcing data integrity and our example queries from the application use joins. Now what? We have three options:

1. modify your data model, reducing data normalization and eliminating constraints, meaning data may have to be duplicated on multiple servers
2. modify your code to handle the joins itself, increasing the number of queries each time order data will need to be constructed
3. some combination of the above two

The first steps with our pizza model would be to eliminate the foreign keys on the database tables, which means that the RDBMS can no longer enforce data normalization. After we do that, we need to reduce all of our SELECT statements from our code to only access one table at a time. This will increase the number of queries defined in our pseudocode for Solution 2 from 3 to at least 7. This means that we'll need to do more queries, smaller in individual size, but hopefully faster. Now we can move these tables to almost any machine we wish, allowing us to utilize all this fancy new hardware we just purchased. With those fairly major changes, we can get the system back up and responding.

Solution 3

Server 1
Topping: (id, name, price)
Crust: (id, name, price)
Sauce: (id, name, price)
PizzaLocation: (id, name, priceMultiplier)

Server 2
Pizza: (id, name, crustId, sauceId)
PizzaXTopping: (pizzaId, toppingId, pizzaLocationId)

Server 3
Customer: (id, name, address, phone)

Server 4
Order: (id, customerId, delivery, fulfilled, date)
OrderXPizza: (orderId, pizzaId)

And here's what our queries would look like from the application, and some pseudocode to glue everything together.

order = SELECT * FROM Order WHERE delivery=true AND fulfilled=false
  LIMIT 1 ORDER BY date;
orderCustomer = SELECT * FROM Customer WHERE id=[order.customerId];

pizzaIds = SELECT pizzaId FROM OrderXPizza WHERE orderId = [order.id];
pizzas = SELECT * FROM Pizza WHERE id IN ([pizzaIds]);

foreach pizza in pizzas {
  pizza.displayInfo = SELECT Crust.*, Sauce.* FROM Crust, Sauce
    WHERE Crust.id=[pizza.crustId] AND Sauce.id=[pizza.sauceId];
  pizza.toppings = SELECT * FROM PizzaXTopping WHERE pizzaId=[pizza.id];
  foreach topping in pizza.toppings {
    topping.displayInfo = SELECT Topping.*, PizzaLocation.* FROM
      Topping, PizzaLocation
      WHERE Topping.id=[topping.id] AND PizzaLocation.id=[topping.pizzaLocationId];
  }
}

Small Issues

With the above solution running, we notice that the system is immediately more resilient than it was originally, but it almost immediately begins to slow down as traffic increases. Looking at the database servers, we can see there's excessive load on the Server 1, the system hosting Crust, Sauce, Topping, and PizzaLocation. Digging further, we can see that there are a lot of queries going through for Crust and Sauce, but more than double that amount for Topping and PizzaLocation. A quick eureka moment later, you realize that the data in these tables doesn't change very often, therefore we only really need to request it once and we can reuse it.

Solution 4

This solution uses the same database tables as Solution 3.

allToppings = SELECT * FROM Topping;
allCrusts = SELECT * FROM Crust;
allSauces = SELECT * FROM Sauce;
allPizzaLocations = SELECT * FROM PizzaLocation;

function getToppingById (id) {
  foreach topping in allToppings {
    if topping.id == id { return topping }
  }
}
function getCrustById() { ... }
function getSauceById() { ... }
function getPizzaLocationById { ... }

order = SELECT * FROM Order WHERE delivery=true AND fulfilled=false
  LIMIT 1 ORDER BY date;
orderCustomer = SELECT * FROM Customer WHERE id=[order.customerId];

pizzaIds = SELECT pizzaId FROM OrderXPizza WHERE orderId = [order.id];
pizzas = SELECT * FROM Pizza WHERE id IN ([pizzaIds]);

foreach pizza in pizzas {
  pizza.displayInfo = getCrustById(pizza.crustId) + getSauceById(pizza.sauceId);
  pizza.toppings = SELECT * FROM PizzaXTopping WHERE pizzaId=[pizza.id];
  foreach topping in pizza.toppings {
    topping.displayInfo = getToppingById(topping.id) + getPizzaLocationById(topping.pizzaLocationId);
  }
}

Sizable Issues

With Solution 4 up and running, the system seems to respond much better, and handles the current load, and you finally drift off to a well-deserved sleep.

Is that your PDA buzzing at 3am? Now what? One particular database server has stopped responding... the Pizza server. The size of the Pizza table has exceeded the size of the physical RAM. The RDMBS server put up a good fight to try and keep only relevant records in RAM, but these dang customers just keep coming back and ordering pizza and posting links on their profiles to their pizza-rific creations!

The logical next step is to employ a different technique called table sharding to split up a single table across multiple servers. There are a lot of sharding algorithms out there, most of them don't allow for the number of shards to be increased later on (such as shardId = uniqueKey mod shardCount), so make an effort to pick a good number of shards to start with. We decide to split the Pizza table into 4 shards on 4 different servers. The next step is moving all of the data from one table into 4 different tables, and modifying the code so that it can determine which shard it needs to access. Some of this process can be alleviated if the RDBMS system can handle sharding internally, but most likely, these changes are going to require code changes and an import script.

Once your data is migrated and your code is modified once again, the system is able to grow a bit further.

Solution 5

Database:
Server 1
Topping: (id, name, price)
Crust: (id, name, price)
Sauce: (id, name, price)
PizzaLocation: (id, name, priceMultiplier)

Server 2
PizzaXTopping: (pizzaId, toppingId, pizzaLocationId)

Server 3
Customer: (id, name, address, phone)

Server 4
Order: (id, customerId, delivery, fulfilled, date)
OrderXPizza: (orderId, pizzaId)

Server 5
Pizza_1: (id, name, crustId, sauceId)


Server 6
Pizza_2: (id, name, crustId, sauceId)


Server 7
Pizza_3: (id, name, crustId, sauceId)


Server 8
Pizza_4: (id, name, crustId, sauceId)


Code Change:

allToppings = SELECT * FROM Topping;
allCrusts = SELECT * FROM Crust;
allSauces = SELECT * FROM Sauce;
allPizzaLocations = SELECT * FROM PizzaLocation;

function getToppingById (id) {
  foreach topping in allToppings {
    if topping.id == id { return topping }
  }
}
function getCrustById() { ... }
function getSauceById() { ... }
function getPizzaLocationById { ... }

function getPizzaById(id) {
  shard = id % 4;
  serverId = shard + 4;
  tableName = "Pizza_" + shard;
  pizza = on( serverId, SELECT * FROM [tableName] WHERE id = [id] );
  return pizza;
}

order = SELECT * FROM Order WHERE delivery=true AND fulfilled=false
  LIMIT 1 ORDER BY date;
orderCustomer = SELECT * FROM Customer WHERE id=[order.customerId];

pizzaIds = SELECT pizzaId FROM OrderXPizza WHERE orderId = [order.id];
pizzas = array;
foreach pizzaId in pizzaIds {
  pizzas += getPizzaById( pizzaId );
}

foreach pizza in pizzas {
  pizza.displayInfo = getCrustById(pizza.crustId) + getSauceById(pizza.sauceId);
  pizza.toppings = SELECT * FROM PizzaXTopping WHERE pizzaId=[pizza.id];
  foreach topping in pizza.toppings {
    topping.displayInfo = getToppingById(topping.id) + getPizzaLocationById(topping.pizzaLocationId);
  }
}

Analysis

This approach is essentially how RDBMS systems scale horizontally (ignoring the shortcomings of the code above). There are also methods of clustering, which essentially just abstract out how the sharding is done behind the scenes. Replicas are another weapon in the RDBMS arsenal, but often only serve to increase code complexity and generate race conditions when dealing with huge amounts of interdependent user generated content. In this new web 2.0 frontier, telling Jimmy that "Jesse just bought 2 large cheese pizza's on mypizzaorderingsite.com!" via some social network and giving the next 10 people that click the link a 5% discount increases sales by an astronomical figure. With millions of transactions per hour, speed is no longer a technical requirement, it is a necessity for business survival.

The above solution (focusing on the database) combined with some caching and code ninjutsu can take an application far beyond where most applications will need to go, but it is not without issues. As you can see, the further we go with these examples, the more the focus moves to the code and away from the database.

Next time, we'll look at solutions using non-relational databases.