Sheet 1
Distributed System
  Core factors
    Services
    Redundency
    Partitions
    Handling failure
  Building blocks
    Simple web application
    Assumption: we have TBs of data for random access
      Big data concepts
    Caches
      Principle: LRU
      Limited space but fast
      Can be placed almost anywhere
        Typically on a request layer node
        To scale, each node has its own cache
      Reduce cache misses
        Global caches
          All the nodes use the same single cache space
          When cache miss
            Option1: global cache node retrieves the data
            Option2: the request nodes retrieve the data
        Distributed caches
          Each cache node owns part of the cached data
            Has larger cache space
            Easy to scale up
          Disadvantage
            Remedying a missing node
      Open source option
        Memcached
          Scaling memcached at Facebook
        Many other language- or framework-specific options
    Proxies
      Usage
        Filter requests
        Log requests
        Transform requests
          Add/remove headers
          Encrypt/decrypt
          Compression
      Speed up data access
        Collapsed forwarding
        Collapse spatially close data
      Compare to caches
        Many proxies are also caches
        But not all caches act as proxies
        Generally, caches will be in front of the proxies
      Open source choices
        Squid
        Varnish
    Indexes
      A mapping between the requests and the data location
      Trade-off
        Slower writes, higher storage overhead
        Faster reads
      Application scenario
        TB-size data and small requested payloads (~1kB)
        Create different views of data
          Sorting data
          Inverted indexes
            Intermediate index - Buffer to reduce the index table size
            Nested index architecture
    Load balancers
      Distribute load across a set of nodes
      Algorithms
        Round robin
        Least connection
        Resource based (similar to indexes)
        Fixed weighting
        Source IP hashing
        URL hashing
      Open source choices
        HAProxy
      Hierarchical load balancer in large distributed system
      Reverse proxies
      Challenges
        Manage user-session-specific data
        Make problem diagnosis cumbersome
      Be able to test the health of a node
    Queues
      Build asynchrony into the system
      Provide a strategic abstraction of a client's request and its response
      Provide some protection from service outages and failures
      Open source options
        RabbitMQ
        ActiveMQ
        BeanstalkD
        Zookeeper
        Redis
  Basics
    Service
      Example: Image hosting application
        Successful projects
          Flickr
          Picasa
        Requirements
          Cost effictiency
          High storage scalability
          High availability
          Low latency
        Key parts
          Upload (write) an image to the server
          Query for (read) an image from the server
      Design methodology
        Service-Oriented Architecture (SOA)
          Interact with abstract interface
          Allow each part scales independently
        Cache and persistent storage
          Read from cache
            Use image retrieval service to handle
            Async >500 requests per seconds
          Write to persistent storage
            Use write service to handle
            500 simultaneous writes
        Sharding
          A server or a set of resources serve a subgroup of users
          Drawback: Some operations will be conducted on all shards, such as search or metadata update
    Redundancy
      Data redundancy
      Service redundancy
        Key part: shared-nothing architecture
          Improve resiliency
          Improve scalability
      Failover: switch the traffic to a health copy
    Partitions
      Scale vertically
        Adding more resources to an individual server
      Sacle horizontally
        Adding more nodes
          Take this into account when designing the architecture
        Break up services into partitions/shards
      Find resources
        Use consistent hashing scheme
        Mark resource with incremental ID
      Potential problems
        Inconsistency
        Data locality
    Read more: fault tolerance and monitoring
  Key principles
    Availability
    Performance
    Reliability
    Scalability
    Manageability
    Cost
    Have trade-off
A cache is like short-term memory: it has a limited amount of space, but is typically faster than the original data source and contains the most recently accessed items. Caches can exist at all levels in architecture, but are often found at the level nearest to the front end, where they are implemented to return data quickly without taxing downstream levels.
This involves adding a server, or file store of some sort, faster than your original store and accessible by all the request layer nodes. Each of the request nodes queries the cache in the same way it would a local one. This kind of caching scheme can get a bit complicated because it is very easy to overwhelm a single cache as the number of clients and requests increase, but is very effective in some architectures (particularly ones with specialized hardware that make this global cache very fast, or that have a fixed dataset that needs to be cached).
The majority of applications leveraging global caches tend to use the first type, where the cache itself manages eviction and fetching data to prevent a flood of requests for the same data from the clients.
However, there are some cases where the second implementation makes more sense. For example, if the cache is being used for very large files, a low cache hit percentage would cause the cache buffer to become overwhelmed with cache misses; in this situation it helps to have a large percentage of the total data set (or hot data set) in the cache. Another example is an architecture where the files stored in the cache are static and shouldn't be evicted. (This could be because of application requirements around that data latency—certain pieces of data might need to be very fast for large data sets—where the application logic understands the eviction strategy or hot spots better than the cache.)
Typically the cache is divided up using a consistent hashing function, such that if a request node is looking for a certain piece of data it can quickly know where to look within the distributed cache to determine if that data is available. In this case, each node has a small piece of the cache, and will then send a request to another node for the data before going to the origin. Therefore, one of the advantages of a distributed cache is the increased cache space that can be had just by adding nodes to the request pool.
Some distributed caches get around this by storing multiple copies of the data on different nodes; however, you can imagine how this logic can get complicated quickly, especially when you add or remove nodes from the request layer. Although even if a node disappears and part of the cache is lost, the requests will just pull from the origin—so it isn't necessarily catastrophic!
one function call accessing the cache could make many requests in parallel for data stored on different Memcached servers. This allows them to get much higher performance and throughput for their user profile data, and have one central place to update data (which is important, since cache invalidation and maintaining consistency can be challenging when you are running thousands of servers).
An intermediate piece of hardware/software that receives requests from clients and relays them to the backend origin servers.
collapse the same (or similar) requests together into one request, and then return the single result to the requesting clients.
This is similar to a cache, but instead of storing the data/document like a cache, it is optimizing the requests or calls for those documents and acting as a proxy for those clients.
Employing such a strategy maximizes data locality for the requests, which can result in decreased request latency. We can set up our proxy to recognize the spatial locality of the individual requests, collapsing them into a single request and returning only bigB, greatly minimizing the reads from the data origin.
This can make a really big difference in request time when you are randomly accessing across TBs of data! Proxies are especially helpful under high load situations, or when you have limited caching, since they can essentially batch several requests into one.
It is worth noting that you can use proxies and caches together, but generally it is best to put the cache in front of the proxy, for the same reason that it is best to let the faster runners start first in a crowded marathon race. This is because the cache is serving data from memory, it is very fast, and it doesn't mind multiple requests for the same result. But if the cache was located on the other side of the proxy server, then there would be additional latency with every request before the cache, and this could hinder performance.
These proxy solutions offer many optimizations to make the most of client-server communication. Installing one of these as a reverse proxy (explained in the load balancer section below) at the web server layer can improve web server performance considerably, reducing the amount of work required to handle incoming client requests.
In the case of data sets that are many TBs in size, but with very small payloads (e.g., 1 KB), indexes are a necessity for optimizing data access. Finding a small payload in such a large data set can be a real challenge since you can't possibly iterate over that much data in any reasonable time. Furthermore, it is very likely that such a large data set is spread over several (or many!) physical devices—this means you need some way to find the correct physical location of the desired data. Indexes are the best way to do this.
For large data sets, this is a great way to define different filters and sorts without resorting to creating many additional copies of the data.
For example, imagine that the image hosting system from earlier is actually hosting images of book pages, and the service allows client queries across the text in those images, searching all the book content about a topic, in the same way search engines allow you to search HTML content. In this case, all those book images take many, many servers to store the files, and finding one page to render to the user can be a bit involved. First, inverse indexes to query for arbitrary words and word tuples need to be easily accessible; then there is the challenge of navigating to the exact page and location within that book, and retrieving the right image for the results. So in this case the inverted index would map to a location (such as book B), and then B may contain an index with all the words, locations and number of occurrences in each part.
Creating these intermediate indexes and representing the data in smaller sections makes big data problems tractable. Data can be spread across many servers and still accessed quickly. Indexes are a cornerstone of information retrieval, and the basis for today's modern search engines. Of course, this section only scratched the surface, and there is a lot of research being done on how to make indexes smaller, faster, contain more information (like relevancy), and update seamlessly. (There are some manageability challenges with race conditions, and with the sheer number of updates required to add new data or change existing data, particularly in the event where relevancy or scoring is involved).
Like proxies, some load balancers can also route a request differently depending on the type of request it is.
Make sure the user's content is consistent when they return. If the node is unavailable, there should be ways to recover everything.
Usually, this is done at clients' side, such as browser caches, cookies, and URL rewriting.
For example, when we have sudden high writes, we can put these operations into a queue.
The queue server acknowledges the reception of the requests and will notify the users when it is done.
Imagine the Instagram downloading service. We need to download all photos of a specific user.
For instance, it is quite easy to create a highly robust queue that can retry service requests that have failed due to transient server failures. It is more preferable to use a queue to enforce quality-of-service guarantees than to expose clients directly to intermittent service outages, requiring complicated and often-inconsistent client-side error handling.
Each node is able to operate independently of one another and there is no central "brain" managing state or coordinating activities for the other nodes.
So for a very large data set, this might mean adding more (or bigger) hard drives so a single server can contain the entire data set. In the case of the compute operation, this could mean moving the computation to a bigger server with a faster CPU or more memory. In each case, vertical scaling is accomplished by making the individual resource capable of handling more on its own.
In the case of the large data set, this might be a second server to store parts of the data set, and for the computing resource it would mean splitting the operation or load across some additional nodes. To take full advantage of horizontal scaling, it should be included as an intrinsic design principle of the system architecture, otherwise it can be quite cumbersome to modify and separate out the context to make this possible.
The partitions can be distributed such that each logical set of functionality is separate; this could be done by geographic boundaries, or by another criteria like non-paying versus paying users. The advantage of these schemes is that they provide a service or data store with added capacity.
When there are different services reading and writing from a shared resource, potentially another service or data store, there is the chance for race conditions—where some data is supposed to be updated, but the read happens prior to the update—and in those cases the data is inconsistent.
In distributed systems the closer the data to the operation or point of computation, the better the performance of the system. Therefore it is potentially problematic to have data spread across multiple servers, as any time it is needed it may not be local, forcing the servers to perform a costly fetch of the required information across the network.