Scalable System Design

Building scalable system is becoming a hotter and hotter topic. Mainly because more and more people are using computer these days, both the transaction volume and their performance expectation has grown tremendously.

This one covers general considerations. I have another blogs with more specific coverage on DB scalability as well as Web site scalability.

General Principles

"Scalability" is not equivalent to "Raw Performance"

• Scalability is about reducing the adverse impact due to growth on performance, cost, maintainability and many other aspects
• e.g. Running every components in one box will have higher performance when the load is small. But it is not scalable because performance drops drastically when the load is increased beyond the machine's capacity
  

Understand environmental workload conditions that the system is design for

• Dimension of growth and growth rate: e.g. Number of users, Transaction volume, Data volume
• Measurement and their target: e.g. Response time, Throughput

Understand who is your priority customers

• Rank the importance of traffic so you know what to sacrifice in case you cannot handle all of them

Scale out and Not scale up

• Scale the system horizontally (adding more cheap machine), but not vertically (upgrade to a more powerful machine)

Keep your code modular and simple

• The ability to swap out old code and replace with new code without worries of breaking other parts of the system allows you to experiment different ways of optimization quickly
• Never sacrifice code modularity for any (including performance-related) reasons
  

Don't guess the bottleneck, Measure it

• Bottlenecks are slow code which are frequently executed. Don't optimize slow code if they are rarely executed
• Write performance unit test so you can collect fine grain performance data at the component level
  
• Setup a performance lab so you can conduct end-to-end performance improvement measurement easily

Plan for growth

• Do regular capacity planning. Collect usage statistics, predict the growth rate

Common Techniques

Server Farm (real time access)

• If there is a large number of independent (potentially concurrent) request, then you can use a server farm which is basically a set of identically configured machine, frontend by a load balancer.
• The application itself need to be stateless so the request can be dispatched purely based on load conditions and not other factors.
  
• Incoming requests will be dispatched by the load balancer to different machines and hence the workload is spread and shared across the servers in the farm.
• The architecture allows horizontal growth so when the workload increases, you can just add more server instances into the farm.
  
• This strategy is even more effective when combining with Cloud computing as adding more VM instances into the farm is just an API call.

Data Partitioning

• Spread your data into multiple DB so that data access workload can be distributed across multiple servers
• By nature, data is stateful. So there must be a deterministic mechanism to dispatch data request to the server that host the data
• Data partitioning mechanism also need to take into considerations the data access pattern. Data that need to be accessed together should be staying in the same server. A more sophisticated approach can migrate data continuously according to data access pattern shift.
• Most distributed key/value store do this

Map / Reduce (Batch Parallel Processing)

• The algorithm itself need to be parallelizable. This usually mean the steps of execution should be relatively independent of each other.
• Google's Map/Reduce is a good framework for this model. There is also an open source Java framework Hadoop as well.

Content Delivery Network (Static Cache)

• This is common for static media content. The idea is to create many copies of contents that are distributed geographically across servers.
• User request will be routed to the server replica with close proxmity

Cache Engine (Dynamic Cache)

• This is a time vs space tradeoff. Some executions may use the same set of input parameters over and over again. Therefore, instead of redo the same execution for same input parameters, we can remember the previous execution's result.
• This is typically implemented as a lookup cache.
• Memcached and EHCache are some of the popular caching packages

Resources Pool

• DBSession and TCP connection are expensive to create, so reuse them across multiple requests

Calculate an approximate result

• Instead of calculate an accurate answer, see if you can tradeoff some accuracy for speed.
• If real life, usually some degree of inaccuracy is tolerable

Filtering at the source

• Try to do more processing upstream (where data get generated) than downstream because it reduce the amount of data being propagated

Asynchronous Processing

• You make a call which returns a result. But you don't need to use the result until at a much later stage of your process. Therefore, you don't need to wait immediately after making the call., instead you can proceed to do other things until you reach the point where you need to use the result.
• In additional, the waiting thread is idle but consume system resources. For high transaction volume, the number of idle threads is (arrival_rate * processing_time) which can be a very big number if the arrival_rate is high. The system is running under a very ineffective mode
  
• The service call in this example is better handled using an asynchronous processing model. This is typically done in 2 ways: Callback and Polling
• In callback mode, the caller need to provide a response handler when making the call. The call itself will return immediately before the actually work is done at the server side. When the work is done later, response will be coming back as a separate thread which will execute the previous registered response handler. Some kind of co-ordination may be required between the calling thread and the callback thread.
• In polling mode, the call itself will return a "future" handle immediately. The caller can go off doing other things and later poll the "future" handle to see if the response if ready. In this model, there is no extra thread being created so no extra thread co-ordination is needed.

Implementation design considerations

• Use efficient algorithms and data structure. Analyze the time (CPU) and space (memory) complexity for logic that are execute frequently (ie: hot spots). For example, carefully decide if hash table or binary tree should be use for lookup.
• Analyze your concurrent access scenarios when multiple threads accessing shared data. Carefully analyze the synchronization scenario and make sure the locking is fine-grain enough. Also watch for any possibility of deadlock situation and how you detect or prevent them. A wrong concurrent access model can have huge impact in your system's scalability. Also consider using Lock-Free data structure (e.g. Java's Concurrent Package have a couple of them)
• Analyze the memory usage patterns in your logic. Determine where new objects are created and where they are eligible for garbage collection. Be aware of the creation of a lot of short-lived temporary objects as they will put a high load on the Garbage Collector.
  
• However, never trade off code readability for performance. (e.g. Don't try to bundle too much logic into a single method). Let the VM handle this execution for you.

Classification via Decision Tree

Decision Tree is another model-based learning approach where the output is a tree. Here, we are given a set of data with structure [x1, x2 …, y] is presented. (in this case y is the output). The learning algorithm will learn (from the training set) a decision tree and use that to predict the output y for future seen data.

x1, x2 ... can be either numeric or categorical. y1 is categorical

In the decision tree, the leaf node contains relatively "pure" data represented by a histogram {class_j => count}. The intermediate node is called a "decision node" containing a test against an input attribute (e.g. x2) to a constant value. Two branches are output according to whether the decision is evaluate to be true or false. If x2 is a categorical value, the test will be a equality test (e.g. if "weather" equals "sunny"). If x2 is numeric, the test will be a greater than / less than test (e.g. if "age" >= 40).

Building the Decision Tree

We start at the root node which contains all training samples. We need to figure out what should be our first test. Our strategy is to pick the test such that it divides the training samples into two groups which has the highest sum of "purity"

A set of data records is "pure" if all their outcome is gravitated towards a particular value, otherwise it is impure. Purity can be measurable by Entropy or Gini Impurity function.

Gini is measured by calculating the probability of picking two records from a set such that their outcome is different.

Entropy is measured by calculate the following ...
sum_over_j(P(class_j) * log (P(class_j)))

Note that the term P * logP is close to zero in either case when P is close to zero or when P is close to one. Entropy is large when P is about 0.5. The higher the entropy, the lower the purity.

Keep doing the following until the overall purity is not improved further

1. Try all combination of x1, x2 ... / value1a, value1b, value2a, value2b ...
   
2. Pick one combination such that the divided data set has a better combined purity (which is the weighted sum of purity based on the frequency)
3. To avoid the decision tree overfits the training data, we divide the tree only when the purity after divide exceed a threshold value (called pre-pruning)
   

def build_tree(set, func, threshold)
 orig_purity = calculate_purity(func, set)
 purity_gain = 0
 best_attribute = nil
 best_test_value = nil
 best_split_left_set = nil
 best_split_right_set = nil

 for x in each attribute
   for x_value in each possible value of x
     left_set, right_set = split(x, x_value, set)
     left_freq = left_set.size / set.size
     right_freq = right_set.size / set.size
     left_purity = calculate_purity(func, left_set)
     right_purity = calculate_purity(func, right_set)
     split_purity = left_freq*left_purity + right_freq*right_purity
     improvement =  split_purity - orig_purity
     if improvement > purity_gain
       purity_gain = improvement
       best_attribute = x
       best_test_value = x_value
       best_split_left_set = left_set
       best_split_right_set = right_set

 if purity_gain > threshold
   node = DecisionNode.new(best_attr, best_value)
   node.left = build_tree(best_fit_left_set, func, threshold)
   node.right = build_tree(best_fit_right_set, func, threshold)
 else
   node = DecisionNode.new
   node.result = calculate_outcome_freq(set)

Over Fitting and Tree Pruning

Since the training data set is not exactly same as the actual population, we may biased towards some characteristics of the training data set which doesn't exist in actual population. Such biases is called over fitting and we want to avoid it.

In above algorithm, we create an intermediate node when the purity_gain is significant. We can also do "post pruning" such that we build the tree first and then working backward trying to merge the leaf nodes if the decrease of purity is small.

We also can set aside 1/4 of training data as validation data. So we use the 3/4 of training data to build the tree first and then use the 1/4 validation data to prune the tree. This approach will reduce the training data size and so only practical when the training data set is large.


Classification and missing data handling

When a new query point arise, we start traversing the decision tree from its root by answering the question at each decision node until we reach the leaf node, from there we pick the class with the highest number of instance count.

However, what if the value of some attributes (say x3, x5) are missing. When we reach a decision node where x3 need to be tested, we cannot proceed.

One simple solution is to fill-in the most likely value of x3 based on the probability distribution of x3 within the training set. We can also throw a dice based on the probability distribution. Or we can use a tree-way branch by adding "missing data" as one of the condition.

Another more sophisticated solution is to walk down both path if the attributed under test is missed. We know the probability distribution at the junction point, which used to calculate the weight on each branch. The final result will be aggregated according to the weight.

Solving algorithmic problem

There are some general techniques that I've found quite useful to come up with a solution for an algorithmic problem.

Understand the problem

1. Make sure the problem is well defined and understood. Try to rephrase the problem in different ways to double verify the understanding is correct.
2. Understand the constraints and optimization goals of the expected solution. There will be many conflicting dimensions of a solution (e.g. performance, security, scalability ... etc), knowing what can be traded off is important.

Design the solution

1. While you are thinking about the solution, try to make it visual. e.g. Sketch the solution on a whiteboard to help you think.
   
2. See if you can solve the problem by using a simple-minded, brute force search. Try all the possible combinations of solution to determine which one works. But get a sense of the BigO complexity analysis to judge whether the brute-force approach is feasible.
3. See if you translate the problem into a search problem. Then you can utilize existing search algorithm (e.g. hill climbing, A*, BFS, DFS ... etc) to help.
   
4. By exploit the nature of the problem, see if you can define your solution recursively. Assume that you already have a solution with the problem of the smaller size, can you use that to construct the current solution. e.g. Can you define F(n) based on F(n-1), F(n-2) .... etc. And also try to see if you can translate the recursive solution into an iterative one.
5. At this point, you may have found multiple solutions. Do a BigO analysis in both time and space and pick the one that aligns with your optimization goal.

Verify the solution

1. Walk through some simple examples to test out your solution, include some boundary cases in the examples to see how your solution handle those corner cases.
2. Prototype your solution. Define some measurement and instrument your prototype. Run it and compare the result with what you expect.

Recursion vs Iteration

Recursion is a very commonly used technique to express mathematical relationship as well as implement computer algorithm. Define a function recursively usually result in very concise form and ease the understanding. However, recursive function is not performance friendly in terms of stack growth and processing time.

Here is a general form of a recursive function ...

def f(n)
  if (n <= j)
    return known_value[n]
  else
    return g(f(n-1), f(n-2), ... f(n-j))

When this function is invoked, the stack size will grow linearly O(n) and the processing time will grow exponentially O(j**n)

However, the above function can be rewritten into an iteration form. The idea is to revert the order of execution, moving from the known points to the unknown values.

def f(n)
  if (n <= j)
    return known_value[n]
  for i in 0..j
    cache[i] = known_value[i]
  for i in (j+1) .. n
    cache[i] = g(cache[i-1], cache[i-2], ... cache[i-j])
  return cache[n]

In this iteration form, the stack size will not grow and the processing time will grow linearly O(n). Therefore, performance gains significantly when the recursive form is rewrittened in the iteration form.

A* Search

Given a start state S and a Goal state G, find an optimal path S -> A -> B -> ... -> G

We can assign a cost to each arc so the optimal path is represented with the lowest sum of cost. This problem can be modeled as a weighted Graph where nodes represent a state and cost associated arcs represent the transition from one state to another.

A* search is one of the most popular search algorithm and it has the following characteristics

1. If there is a path existing between S to G, it can always find one
2. It will always find the one that is optimal (least cost)
   

Here is the description of the algorithm

• Define a function: f(X) = g(X) + h(X) where ...
• g(X) = minimal cost path from A to X
• h(X) = a heuristic estimation cost from X to S. It is important that the estimated cost h(X) is less than the actual cost
  

1. The algorithm keep track of a list of partially explored path, with an associated f(X) within a priority queue as well as a list of explored nodes. Initially the priority queue contain just one path [S] with cost C and the visited node list contains just S.
2. Repeat the following ...
3. Pick the lowest cost path ... based on f(X) from the priority queue to explore
4. Set current_node to be the last node of the lowest cost path
5. If current_node is explored already, skip this node and go back to step 3
   
6. If current_node is the goal, then done. This is the optimal path
7. Otherwise, find out all the neighbours of the current_node. For each of them, insert a path that leads to them into the priority queue.
   

def A_Search(S, G)
 visited_nodes = [S]
 priorityQ.add([] << S)
 while priorityQ.non_empty
   lowest_cost_path = priorityQ.first  # according to f(X)
   node_to_explore = lowest_cost_path.last_node
   continue if visited_nodes contains node_to_explore
   if node_to_explore == G
     return lowest_cost_path
   else
     add node_to_explore to visited_nodes
     for each v in node_to_explore.neighbors
       new_path = lowest_cost_path << v
       insert new_path to priorityQ
     end
   end
 end
 return nil # No solution
end

Proof: A* Search can always find a path if it exist

This should be obvious because the while loop will only terminate if the Goal is reached or all connected node has been explored

Proof: A* Search will find the path with least cost

Lets say the algorithm find a path S -> A -> K -> ... -> G which is not the optimal path
S -> A -> B -> ... -> G

That means, in the last extraction from the priorityQ, f(G) is smaller than f(B)

This is not possible because ...
f(G) == cost(S->A->K...G) > cost (S->A->B...G) > g(B) + h(B) == f(B)

Search Engine and Ranking

Crawling

Start from an HTML, save the file, crawl its links

def collect_doc(doc)
-- save doc
-- for each link in doc
---- collect_doc(link.href)

Build Word Index

Build a set of tuples

{
word1 => {doc1 => [pos1, pos2, pos3], doc2 => [pos4]}
word2 => {...}
}

def build_index(doc)
-- for each word in doc.extract_words
---- index[word][doc] << style="font-weight: bold;" size="5">Search

Given a set of words, locate the relevant docs

A simple example ...

def search(query)
-- Find the most selective word within query
-- for each doc in index[most_selective_word].keys
---- if doc.has_all_words(query)
------ result << doc
-- return result


Ranking

There are many scoring functions and the result need to be able to combine these scoring functions in a flexible way

def scoringFunction(query, doc)
-- do something from query and doc
-- return score

Scoring function can be based on word counts, page rank, or where the location of words within the doc ... etc.
Scoring need to be normalized, say within the same range.
weight is between 0 to 1 and the sum of all weights equals to 1

def score(query, weighted_functions, docs)
-- scored_docs = []
-- for each weighted_function in weighted_functions
---- for each doc in docs
------ scored_docs[doc] += (weight_function[:func](query, doc) * weight_function[:weight])
-- return scored_docs.sort(:rank)

Collaborative Filtering

Given a set of user's rating on movies, how to recommend movies to a new user ?

Ideas from the book: Collective Intelligence chapter one

Given a set of ratings per user
[ person1 => {movieA => rating-1A, movieB => rating-1B},
person2 => {movieX => rating-2X, movieY => rating-2Y, movieA => rating-2A}
...
]

Determine similarity

person1 and person2 are similar if they have rate the same movies with similar ratings

person_distance =
square_root of sum of
-- for each movie_name in person1.ratings.keys
---- if (person2.ratings.keys contains movie_name)
------ square(person1.ratings[movie_name] - person1.ratings[movie_name])


Person similarity =
0 if no common movies in corresponding ratings
1 / (1 + person_distance) otherwise

How to find similar persons to personK ?
Calculate every other person's similarity to personK, and sorted by similarity.

How about movie similarity ?

Invert the set of rankings to ...
[ movieA => {person1 => rating-1A, person2 => rating-2A},
movieB => {person1 => rating-1B}
movieX => {person2 => rating-2X}
movieY => {person2 => rating-2Y}
...
]

movie_distance =
square_root of sum of
-- for each person_name in movieX.ratings.keys
---- if (movieX.ratings.keys contains person_name)
------ square(movieX.ratings[person_name] - movieY.ratings[person_name])


Movie similarity =
0 if no common persons in corresponding ratings
1 / (1 + movie_distance) otherwise

How to find similar movies to movieX ?
Calculate every other movie's similarity to movieX, and sorted by similarity.

Making recommendations

Lets say there is a new personK provide his ratings. How do we recommend movies that may interests him ?

User-based filtering

For each person in persons
-- similarity = person_similarity(personK, person)
-- For each movie_name in person.ratings.keys
---- weighted_ratings[movie_name] += (similarity * person.ratings[movie_name])
---- sum_similarity[movie_name] += similarity

For each movie_name in weighted_ratings.keys
-- weighted_ratings[movie_name] /= sum_similarity[movie_name]

return weighted_ratings.sort_by(:rating)


Item-based filtering

Pre-calculate the movie similarity

For each movie in movies
-- neighbors[movie] = rank_similarity(movie, no_of_close_neighbors)

{
movie1 => {movieA => similarity1A, movieB => similarity1B}
movie2 => {movieC => similarity2C, movieD => similarity2D}
}

At run time ...
personK => {movieX => rating-kX, movieY => rating-kY}

For each movie in personK.ratings.keys
-- for each close_movie in neighbors[movie]
---- weight_ratings[close_movie] += neighbors[movie][close_movie] * personK.ratings[movie]
---- similarity_sum[close_movie] += neighbors[movie][close_movie]

For each target_movie in weight_ratings.keys
-- weight_ratings[target_movie] /= similarity_sum[target_movie]

return weight_ratings.sort_by(:rating)