System Design.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Systems Design Fundamentals"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Client - Server Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "The paradigm by which modern systems are designed, which consists of clients requesting data or service from servers and servers providing data or service to clients.\n",
    "\n",
    "### Client\n",
    "A machine or process that requests data or service from a server. \n",
    "\n",
    "### Server\n",
    "A machine or process that provides data or service for a client, usually by listening for incoming network calls.\n",
    "\n",
    "> Note that a single machine or piece of software can be both a client and a server at the same time. For instance, a single machine could act as a server for end users and as a client for a database.\n",
    "\n",
    "### IP Address\n",
    "\n",
    "An address given to each machine connected to the public internet. IPv4 addresses consist of four numbers separated by dots: **a.b.c.d** where all four numbers are between 0 and 255. Special values include:\n",
    "\n",
    "* **127.0.0.1**: Your own local machine, also referred to as **localhost**.\n",
    "* **192.168.x.y**: Your private network. For instance, your machine and all machines on your private wifi network will usually have the **192.168** prefix.\n",
    "\n",
    "### Port\n",
    "\n",
    "In order for multiple programs to listen for new network connections on the same machine without colliding, they pick a port to listen on. A port is an integer between 0 and 65,535 ($2^{16}$ ports total).\n",
    "\n",
    "Typically, ports **0-1023** are reserved for **system ports** (also called **well-known ports**) and shouldn't be used by user-level processes. Higher-numbered ports are available for general use by applications and are known as **ephemeral ports**. Certain ports have pre-defined uses, below are some examples:\n",
    "* 22: Secure Shell\n",
    "* 53: DNS lookup\n",
    "* 80: HTTP\n",
    "* 443: HTTPS\n",
    "\n",
    "\n",
    "### DNS\n",
    "\n",
    "Short for **Domain Name System**, it describes the entities and protocols involved in the translation from domain names to IP Addresses. Typically, machines make a DNS query to a well known entity which is responsible for returning the IP address (or multiple ones) of the requested domain name in the response."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Network Protocols"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### IP\n",
    "\n",
    "Stands for **Internet Protocol**. This network protocol outlines how almost all machine-to-machine communications should happen in the world. Other protocols like **TCP**, **UDP** and **HTTP** are built on top of IP.\n",
    "\n",
    "### TCP\n",
    "\n",
    "Network protocol built on top of the Internet Protocol (IP). Allows for ordered, reliable data delivery between machines over the public internet by creating a connection.\n",
    "\n",
    "The **Transmission Control Protocol** (**TCP**) is usually implemented in the kernel, which exposes **sockets** to applications that they can use to stream data through an open connection.\n",
    "\n",
    "### HTTP\n",
    "\n",
    "The **HyperText Transfer Protocol** is a very common network protocol implemented on top of TCP. Clients make HTTP requests, and servers respond with a response. Following are a sequence of events in a **request-response** mechanism:\n",
    "* The client opens a connection and requests data from the server.\n",
    "* The server calculates the response.\n",
    "* The server sends the response back to the client on the opened request.\n",
    "\n",
    "<img src='imgs/http.png' alt='http' width=500 height=500>\n",
    "\n",
    "Requests typically have the following schema:\n",
    "\n",
    ">* **host**: string (example: algoexpert.io)\n",
    ">* **port**: integer (example: 80 or 443)\n",
    ">* **method**: string (example: GET, PUT, POST, DELETE, OPTIONS or PATCH)\n",
    ">* **headers**: pair list (example: \"Content-Type\" => \"application/json\")\n",
    ">* **body**: opaque sequence of bytes\n",
    "\n",
    "Responses typically have the following schema:\n",
    ">* **status code**: integer (example: 200, 401)\n",
    ">* **headers**: pair list (example: \"Content-Length\" -> 1238)\n",
    ">* **body**: opaque sequence of bytes\n",
    "\n",
    "### IP Packet\n",
    "\n",
    "Sometimes more broadly referred to as just a (network) **packet**, an IP packet is effectively the smallest unit used to describe data being sent over IP, aside from bytes. An IP packet consists of:\n",
    "\n",
    "* an **IP header**, which contains the source and destination IP addresses as well as other information related to the network\n",
    "* a **payload**, which is just the data being sent over the network\n",
    "\n",
    "### Socket\n",
    "A socket is one endpoint of a two-way communication link between two programs running on the network. A socket is bound to a port number so that the **TCP layer** can identify the application that data is destined to be sent to. An endpoint is a \n",
    "\n",
    "A socket is a combination of an **IP address** and a **port number** and is used to identify both a machine and a service within the machine."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Storage"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### Databases\n",
    "\n",
    "Databases are programs that either use disk or memory to do 2 core things: **record** data and **query** data. In general, they are: themselves servers that are long lived and interact with the rest of our application through network calls, with protocols on top of TCP or even HTTP.\n",
    "\n",
    "Some databases only keep records in memory, and the users of such databases are aware of the fact that those records may be lost forever if the machine or process dies.\n",
    "\n",
    "For the most part though, databases need persistence of those records, and thus cannot use memory. This means that we have to write our data to disk. Anything written to disk will remain through power loss or network partitions, so that's what is used to keep permanent records.\n",
    "\n",
    "Since machines die often in a large scale system, special disk partitions or volumes are used by the database processes, and those volumes can get recovered even if the machine were to go down permanently.\n",
    "\n",
    "### Disk\n",
    "\n",
    "Usually refers to either **HDD** (**hard-disk drive**) or **SSD** (**solid-state drive**). Data written to disk will persist through power failures and general machine crashes. Disk is also referred to as **non-volatile storage**.\n",
    "\n",
    "SSD is far faster than HDD but also far more expensive from a financial point of view. Because of that, HDD will typically be used for data that's rarely accessed or updated, but that's stored for a long time, and SSD will be used for data that's frequently accessed and updated.\n",
    "\n",
    "### Memory\n",
    "\n",
    "Short for **Random Access Memory** (**RAM**). Data stored in memory will be lost when the process that has written that data dies.\n",
    "\n",
    "### Persistent Storage\n",
    "\n",
    "Usually refers to disk, but in general it is any form of storage that persists if the process in charge of managing it dies.\n",
    "\n",
    "\n",
    "### Storage Concepts\n",
    "There are three different storage concepts - **File Storage**, **Block Storage** and **Object Storage**.\n",
    "\n",
    "#### File storage\n",
    "* **File storage**, also known as **hierarchical storage**, is the way to access files on a personal computer, as individual items stored in directories or folders. We can access data via the hierarchical system, i.e. files are referenced by the directory/folder path.\n",
    "\n",
    "#### Block Storage\n",
    "* **Block Storage** breaks data up into blocks, each block of data has a unique identifier and each block can be in a different location. Accessing Block storage is faster because we do not have to trace down a file tree. Block Storage is also file system independent and the data can be spread across devices running different file systems. \n",
    "\n",
    "\n",
    "* The block device refers to the particular piece of hardware that is used to store data. Examples of block devices are: hard drives, solid state drives, flash drives, Storage Area Networks (SAN) and cloud storage.\n",
    "\n",
    "\n",
    "* The advantages of block storage is that it allows data to be spread over multiple environments, storage is not tied to file system compatibility. Block storage data retrieval is also faster than file storage data retrieval because there can be multiple paths to a block.\n",
    "\n",
    "\n",
    "* A disadvantage of block storage is that there is no metadata and therefore not as suitable for unstructured data. Search capabilities are limited with this storage format.\n",
    "\n",
    "\n",
    "#### Object storage\n",
    "* **Object storage**, also referred to as **object based storage**, is the newest technology of these three. With Object storage the application does not need to know where the data resides. Object data is retrieved through **APIs** using the unique identifier associated with the object.\n",
    "\n",
    "\n",
    "* In the Object Storage model, each object includes the data, metadata and a unique identifier. This method is particularly well suited for storing large amounts of unstructured data that is not being continuously modified (static data) such as videos, photos and songs.\n",
    "\n",
    "\n",
    "* Typically Object-based storage uses **TCP/IP** as its transport, and devices communicate using **HTTP** and **REST APIs**.\n",
    "\n",
    "\n",
    "* **Object Storage** is the cheapest of these three types of storage and most cost effective for high volumes of data. **File storage** works best for low volumes of data."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### FUSE"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "**FUSE** (**Filesystem in Userspace**) allows users to create their own file systems without editing Linux kernel code. These file systems may be mounted by users who do not have root privileges. \n",
    "\n",
    "We can use FUSE to extend the features of the Linux file system without the need to learn Linux file system internals or kernel module programming.\n",
    "\n",
    "The libfuse library offers two Application Programmer Interfaces (**APIs**), the high level API allows us to access files using **file names** and **paths** and the low level API uses **inodes**."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Databases"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### Relational Database\n",
    "A type of **structured database** in which data is stored following a **tabular** format; often supports powerful querying using SQL.\n",
    "\n",
    "Each **row** contains all the information about one entity and each **column** contains all the separate data points. Some of the most popular relational databases are MySQL, Oracle, MS SQL Server, SQLite, Postgres, and MariaDB.\n",
    "\n",
    "In relational databases, each record conforms to a **fixed schema**, meaning the columns must be decided and chosen before data entry and each row must have data for each column. The schema can be altered later, but it involves modifying the whole database and going offline.\n",
    "\n",
    "In most common situations, SQL databases are **vertically scalable**, i.e., by increasing the horsepower (higher Memory, CPU, etc.) of the hardware, which can get very expensive. It is possible to scale a relational database across multiple servers, but this is a challenging and time-consuming process.\n",
    "\n",
    "### Non-Relational Database\n",
    "In contrast with relational database (SQL databases), a type of database that is free of imposed, tabular-like structure. Non-relational databases are often referred to as **NoSQL databases**.\n",
    "\n",
    "In NoSQL, schemas are **dynamic**. Columns can be added on the fly and each `row` (or equivalent) doesn’t have to contain data for each `column`.\n",
    "\n",
    "NoSQL databases are **horizontally scalable**, meaning we can add more servers easily in our NoSQL database infrastructure to handle a lot of traffic. Any cheap commodity hardware or cloud instances can host NoSQL databases, thus making it a lot more cost-effective than vertical scaling. A lot of NoSQL technologies also distribute data across servers automatically.\n",
    "\n",
    "### SQL\n",
    "**Structured Query Language**. Relational databases can be used using a derivative of SQL such as PostgreSQL in the case of Postgres.\n",
    "\n",
    "### SQL Database\n",
    "\n",
    "Any database that supports SQL. This term is often used synonymously with **Relational Database**, though in practice, not every relational database supports SQL.\n",
    "\n",
    "### NoSQL Database\n",
    "\n",
    "Any database that is not SQL-compatible is called **NoSQL**.\n",
    "\n",
    "### ACID Transaction\n",
    "A type of database transaction that has four important properties:\n",
    "\n",
    "* **Atomicity**: The operations that constitute the transaction will either all succeed or all fail. There is no in-between state.\n",
    "* **Consistency**: The transaction cannot bring the database to an invalid state. After the transaction is committed or rolled back, the rules for each record will still apply, and all future transactions will see the effect of the transaction. Also named **Strong Consistency**.\n",
    "* **Isolation**: The execution of multiple transactions concurrently will have the same effect as if they had been executed sequentially.\n",
    "* **Durability**: Any committed transaction is written to non-volatile storage. It will not be undone by a crash, power loss, or network partition.\n",
    "\n",
    "The vast majority of relational databases are **ACID compliant**. So, when it comes to data reliability and safe guarantee of performing transactions, SQL databases are still the better bet.\n",
    "\n",
    "Most of the NoSQL solutions sacrifice ACID compliance for performance and scalability.\n",
    "\n",
    "### Database Index\n",
    "\n",
    "A special auxiliary **data structure** that allows our database to perform certain queries much faster. Indexes can typically only exist to reference structured data, like data stored in relational databases. In practice, we create an index on one or multiple columns in our database to greatly speed up read queries that we run very often, with the downside of slightly longer writes to our database, since writes have to also take place in the relevant index.\n",
    "\n",
    "An **index** is a data structure that can be perceived as a table of contents that points us to the location where actual data lives. So when we create an index on a column of a table, we store that column and a pointer to the whole row in the index. Let’s assume a table containing a list of books, the following diagram shows how an index on the `Title` column looks like:\n",
    "\n",
    "<img src='imgs/index.png' alt='index' width=500 height=500>\n",
    "\n",
    "Just like a traditional relational data store, we can also apply this concept to larger datasets. The trick with indexes is that we must carefully consider how users will access the data. In the case of data sets that are many terabytes in size, but have very **small payloads** (e.g., 1 KB), indexes are a necessity for optimizing data access. Finding a small payload in such a large dataset can be a real challenge, since we 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 physical devices—this means we need some way to find the correct physical location of the desired data. Indexes are the best way to do this.\n",
    "\n",
    "#### Indexes decrease write performance\n",
    "An index can dramatically speed up data retrieval but may itself be large due to the additional keys, which slow down data insertion & update.\n",
    "\n",
    "When adding rows or making updates to existing rows for a table with an active index, we not only have to write the data but also have to update the index. This will decrease the write performance. This performance degradation applies to all insert, update, and delete operations for the table. For this reason, adding unnecessary indexes on tables should be avoided and indexes that are no longer used should be removed. To reiterate, adding indexes is about improving the performance of search queries. If the goal of the database is to provide a data store that is often written to and rarely read from, in that case, decreasing the performance of the more common operation, which is writing, is probably not worth the increase in performance we get from reading.\n",
    "\n",
    "### Strong Consistency\n",
    "\n",
    "Strong Consistency usually refers to the consistency of ACID transactions, as opposed to Eventual Consistency.\n",
    "\n",
    "### Eventual Consistency\n",
    "\n",
    "A consistency model which is unlike Strong Consistency. In this model, reads might return a view of the system that is stale. An eventually consistent datastore will give guarantees that the state of the database will eventually reflect writes within a time period (could be 10 seconds, or minutes).\n",
    "\n",
    "### Postgres\n",
    "\n",
    "A relational database that uses a dialect of SQL called PostgreSQL. Provides ACID transactions.\n",
    "\n",
    "### Database Lock\n",
    "\n",
    "In a relational database that provides ACID transactions, updating rows inside a table will cause a **lock** to be held on that table or on the rows we are updating. If a second transaction tries to update the same rows, it will block before the update until the first transaction releases that lock. This is one of the core mechanisms behind the **Atomicity** of ACID transactions.\n",
    "\n",
    "### Database Selection\n",
    "Here are a few reasons to choose a **SQL database**:\n",
    "\n",
    "* We need to ensure **ACID compliance**. ACID compliance reduces anomalies and protects the integrity of our database by prescribing exactly how transactions interact with the database. Generally, NoSQL databases sacrifice ACID compliance for scalability and processing speed, but for many e-commerce and financial applications, an ACID-compliant database remains the preferred option.\n",
    "* Our data is **structured** and **unchanging**. If our business is not experiencing massive growth that would require more servers and if we’re only working with data that is **consistent**, then there may be no reason to use a system designed to support a variety of data types and high traffic volume.\n",
    "\n",
    "Reasons to use **NoSQL database** are:\n",
    "* When all the other components of our application are fast and seamless, NoSQL databases prevent data from being the bottleneck. Big data is contributing to a large success for NoSQL databases, mainly because it handles data differently than the traditional relational databases. A few popular examples of NoSQL databases are MongoDB, CouchDB, Cassandra, and HBase.\n",
    "* Storing large volumes of data that often have little to no structure. A NoSQL database sets no limits on the types of data we can store together and allows us to add new types as the need changes. With document-based databases, we can store data in one place without having to define what `types` of data those are in advance.\n",
    "* Making the most of cloud computing and storage. Cloud-based storage is an excellent cost-saving solution but requires data to be easily spread across multiple servers to scale up. Using commodity (affordable, smaller) hardware on-site or in the cloud saves you the hassle of additional software and NoSQL databases like Cassandra are designed to be scaled across multiple data centers out of the box, without a lot of headaches.\n",
    "* Rapid development. NoSQL is extremely useful for rapid development as it doesn’t need to be prepped ahead of time. If we’re working on quick iterations of our system which require making frequent updates to the data structure without a lot of downtime between versions, a relational database will slow us down."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Stateful and Stateless"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### Stateful\n",
    "\n",
    "A server or process is called **stateful** when it derives its functionality from storing and retrieving things from disk. Databases are primary case studies for stateful servers. Because of this persistence requirement, it's much more difficult to run and manage stateful servers compared to **Stateless** servers because they can't be stopped and restarted on any physical machine.\n",
    "\n",
    "### Stateless\n",
    "\n",
    "A server is usually called **stateless** if it does not require state to be persisted to disk in order to run successfully. Although many server process typically hold some state in memory including caching layers for instance, this typically means that we can run the server process the same way on any machine, and move it around whenever we want. This contrasts with **Stateful** processes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Latency And Throughput"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### Latency\n",
    "\n",
    "The time it takes for a certain operation to complete in a system. Most often this measure is a time duration, like milliseconds or seconds. You should know these orders of magnitude:\n",
    "\n",
    "* **Reading 1 MB from RAM**: 250 μs (0.25 ms)\n",
    "* **Reading 1 MB from SSD**: 1,000 μs (1 ms)\n",
    "* **Transfer 1 MB over Network**: 10,000 ps (10 ms)\n",
    "* **Reading 1MB from HDD**: 20,000 μs (20 ms)\n",
    "* **Inter-Continental Round Trip**: 150,000 μs (150 ms)\n",
    "\n",
    "### Throughput\n",
    "\n",
    "The number of operations that a system can handle properly per time unit. For instance the throughput of a server can often be measured in requests per second (RPS or QPS)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Availability"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Process\n",
    "\n",
    "A program that is currently running on a machine. We should always assume that any process may get terminated at any time in a sufficiently large system.\n",
    "\n",
    "### Node/ Instance/ Host\n",
    "\n",
    "These three terms refer to the same thing most of the time: a **virtual or physical machine** on which the developer runs processes. Sometimes the word server also refers to this same concept.\n",
    "\n",
    "### Availability\n",
    "\n",
    "The odds of a particular server or service being up and running at any point in time, usually measured in percentages. A server that has 99% availability will be operational 999% of the time (this would be described as having two nines of availability).\n",
    "\n",
    "### High Availability\n",
    "\n",
    "Used to describe systems that have particularly high levels of availability, typically 5 nines or more: sometimes abbreviated **HA**.\n",
    "\n",
    "### Nines\n",
    "\n",
    "Typically refers to percentages of **uptime**. For example, 5 nines of availability means an uptime of 99.999% of the time. Below are the downtimes expected per year depending on those 9s\n",
    "\n",
    ">* **99% (two 9s)**: 87.7 hours\n",
    ">* **99.9% (three 9s)**: 8.8 hours\n",
    ">* **99.99%**: 52.6 minutes\n",
    ">* **99.999%**: 5.3 minutes\n",
    "\n",
    "### Redundancy\n",
    "\n",
    "The process of replicating parts of a system in an effort to make it more reliable.\n",
    "\n",
    "### SLA\n",
    "\n",
    "Short for **service-level agreement**, an SLA is a collection of guarantees given to a customer by a service provider. SLAs typically make guarantees on a system's availability, amongst other things. SLAS are made up of one or multiple SLOs.\n",
    "\n",
    "### SLO\n",
    "\n",
    "Short for **service-level objective**, an SLO is a guarantee given to a customer by a service provider. SLOs typically make guarantees on a system's availability, amongst other things. SLOs constitute an SLA.\n",
    "\n",
    "### Availability Zone\n",
    "\n",
    "Sometimes referred to as an **AZ**, an availability zone designates a group of machines that share one or more central system components (e.g., power source, network connectivity, machine-cooling system). Availability zones are typically located far away from each other such that no natural disaster can realistically bring down two of them at once. This ensures that if we have redundant storage, for instance, with data stored in two availability zones, losing one AZ still leaves us with an operational system that abides by any **SLA** that it might have.\n",
    "\n",
    "### Percentiles\n",
    "\n",
    "Most often used when describing a **latency distribution**. If our **X**th percentile is 100 milliseconds, it means that **X**% of the requests have latencies of 100ms or less. Sometimes, SLAs describe their guarantees using these percentiles."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Caching"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### Cache\n",
    "\n",
    "A piece of hardware or software that stores data, typically meant to retrieve that data faster than otherwise. Caches take advantage of the locality of reference principle: recently requested data is likely to be requested again. They are used in almost every computing layer: hardware, operating systems, web browsers, web applications, and more. 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.\n",
    "\n",
    "Caches are often used to store responses to network requests as well as results of computationally-long operations.\n",
    "\n",
    "Note that data in a cache can become **stale** if the main source of truth for that data (ie, the main database behind the cache) gets updated and the cache doesn't.\n",
    "\n",
    "\n",
    "### Cache Hit\n",
    "\n",
    "When requested data is found in a cache.\n",
    "\n",
    "### Cache Miss\n",
    "\n",
    "When requested data could have been found in a cache but isn't. This is typically used to refer to a negative consequence of a system failure or of a poor design choice. For example:\n",
    "\n",
    "If a server goes down, our load balancer will have to forward requests to a new server, which will result in cache misses.\n",
    "\n",
    "\n",
    "### Cache Invalidation\n",
    "While caching is fantastic, it requires some maintenance to keep the cache coherent with the source of truth (e.g., database). If the data is modified in the database, it should be invalidated in the cache; if not, this can cause inconsistent application behavior.\n",
    "\n",
    "Solving this problem is known as cache invalidation; there are three main schemes that are used:\n",
    "\n",
    "**Write-through cache**: Under this scheme, data is written into the cache and the corresponding database simultaneously. The cached data allows for fast retrieval and, since the same data gets written in the permanent storage, we will have complete data consistency between the cache and the storage. Also, this scheme ensures that nothing will get lost in case of a crash, power failure, or other system disruptions.\n",
    "\n",
    "Although, write-through minimizes the risk of data loss, since every write operation must be done twice before returning success to the client, this scheme has the disadvantage of higher latency for write operations.\n",
    "\n",
    "**Write-around cache**: This technique is similar to write-through cache, but data is written directly to permanent storage, bypassing the cache. This can reduce the cache being flooded with write operations that will not subsequently be re-read, but has the disadvantage that a read request for recently written data will create a “cache miss” and must be read from slower back-end storage and experience higher latency.\n",
    "\n",
    "**Write-back cache**: Under this scheme, data is written to cache alone, and completion is immediately confirmed to the client. The write to the permanent storage is done after specified intervals or under certain conditions. This results in low-latency and high-throughput for write-intensive applications; however, this speed comes with the risk of data loss in case of a crash or other adverse event because the only copy of the written data is in the cache.\n",
    "\n",
    "\n",
    "### Cache Eviction Policy\n",
    "\n",
    "The policy by which values get evicted or removed from a cache. Popular cache eviction policies include **LRU** (**least-recently used**), **FIFO** (**first in first out**), and **LFU** (**least-frequently used**). Following are some of the most common cache eviction policies:\n",
    "\n",
    "* **First In First Out (FIFO)**: The cache evicts the first block accessed first without any regard to how often or how many times it was accessed before.\n",
    "* **Last In First Out (LIFO)**: The cache evicts the block accessed most recently first without any regard to how often or how many times it was accessed before.\n",
    "* **Least Recently Used (LRU)**: Discards the least recently used items first.\n",
    "* **Most Recently Used (MRU)**: Discards, in contrast to LRU, the most recently used items first.\n",
    "* **Least Frequently Used (LFU)**: Counts how often an item is needed. Those that are used least often are discarded first.\n",
    "* **Random Replacement (RR)**: Randomly selects a candidate item and discards it to make space when necessary.\n",
    "\n",
    "### Content Delivery Network\n",
    "\n",
    "A **CDN** is a third-party service that acts like a cache for our servers. Sometimes, web applications can be slow for users in a particular region if our servers are located only in another region. A CDN has servers all around the world, meaning that the latency to a CDN's servers will almost always be far better than the latency to our servers. A CDN's servers are often referred to as **PoPs** (**Points of Presence**). Two of the most popular CDNs are **Cloudflare** and **Google Cloud CDN**."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Hashing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### Hashing Function\n",
    "\n",
    "A function that takes in a specific data type (such as a string or an identifier) and outputs a number. Different inputs may have the same output, but a good hashing function attempts to minimize those **hashing collisions** (which is equivalent to maximizing uniformity).\n",
    "\n",
    "### Consistent Hashing\n",
    "A type of hashing that minimizes the number of keys that need to be remapped when a hash table gets resized. It's often used by load balancers to distribute traffic to servers; it minimizes the number of requests that get forwarded to different servers when new servers are added or when existing servers are brought down.\n",
    "\n",
    "**Distributed Hash Table (DHT)** is one of the fundamental components used in distributed scalable systems. Hash Tables need a key, a value, and a hash function where hash function maps the key to a location where the value is stored.\n",
    "\n",
    "$$index = hash\\_function(key)$$\n",
    "\n",
    "Suppose we are designing a distributed caching system. Given `n` cache servers, an intuitive hash function would be `key % n`. It is simple and commonly used. But it has two major drawbacks:\n",
    "\n",
    "- It is NOT horizontally scalable. Whenever a new cache host is added to the system, all existing mappings are broken. It will be a pain point in maintenance if the caching system contains lots of data. Practically, it becomes difficult to schedule a downtime to update all caching mappings.\n",
    "\n",
    "\n",
    "- It may NOT be load balanced, especially for non-uniformly distributed data. In practice, it can be easily assumed that the data will not be distributed uniformly. For the caching system, it translates into some caches becoming hot and saturated while the others idle and are almost empty.\n",
    "\n",
    "\n",
    "In such situations, consistent hashing is a good way to improve the caching system.\n",
    "\n",
    "**Consistent hashing** is a very useful strategy for distributed caching systems and DHTs. It allows us to distribute data across a cluster in such a way that will minimize reorganization when nodes are added or removed. Hence, the caching system will be easier to scale up or scale down.\n",
    "\n",
    "In Consistent Hashing, when the hash table is resized (e.g. a new cache host is added to the system), only `k/n` keys need to be remapped where `k` is the `total number of keys` and `n` is the `total number of servers`. Recall that in a caching system using the `mod` as the hash function, all keys need to be remapped.\n",
    "\n",
    "In Consistent Hashing, objects are mapped to the same host if possible. When a host is removed from the system, the objects on that host are shared by other hosts; when a new host is added, it takes its share from a few hosts without touching other’s shares.\n",
    "\n",
    "#### Working\n",
    "As a typical hash function, consistent hashing maps a key to an integer. Suppose the output of the hash function is in the range of [0, 256]. Imagine that the integers in the range are placed on a ring such that the values are wrapped around.\n",
    "\n",
    "1. Given a list of cache servers, hash them to integers in the range.\n",
    "2. To map a key to a server\n",
    " * Hash it to a single integer.\n",
    " * Move clockwise on the ring until finding the first cache it encounters, that cache is the one that contains the key. \n",
    "\n",
    "<table>\n",
    "<tr>\n",
    "<td><img src='imgs/h1.png' alt='h1' width=300 height=400 align='left'></td>\n",
    "<td><img src='imgs/h2.png' alt='h2' width=300 height=400 align='middle'></td>\n",
    "<td><img src='imgs/h3.png' alt='h3' width=300 height=400 align='right'></td>\n",
    "<td><img src='imgs/h4.png' alt='h4' width=300 height=400 align='right'></td>\n",
    "<td><img src='imgs/h5.png' alt='h5' width=300 height=400 align='right'></td>\n",
    "</tr></table>\n",
    "\n",
    "\n",
    "- To add a new server, say D, keys that were originally residing at C will be split. Some of them will be shifted to D, while other keys will not be touched.\n",
    "\n",
    "\n",
    "- To remove a cache or, if a cache fails, say A, all keys that were originally mapped to A will fall into B, and only those keys need to be moved to B; other keys will not be affected.\n",
    "\n",
    "For load balancing, the real data is essentially `randomly distributed` and thus may not be uniform. It may make the keys on caches unbalanced. To handle this issue, we add `virtual replicas` for caches. Instead of mapping each cache to a single point on the ring, we map it to multiple points on the ring, i.e. **replicas**. This way, each cache is associated with multiple portions of the ring. \n",
    "\n",
    "If the hash function *mixes well*, **as the number of replicas increases, the keys will be more balanced**.\n",
    "\n",
    "### Rendezvous Hashing\n",
    "\n",
    "A type of hashing also coined **highest random weight hashing**. Allows for minimal re-distribution of mappings when a server goes down.\n",
    "\n",
    "### SHA\n",
    "\n",
    "Short for **Secure Hash Algorithms**, the SHA is a collection of cryptographic hash functions used in the industry. SHA-3 is a popular choice to use in a system.\n",
    "\n",
    "\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Redundancy, Replication And Sharding"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Redundancy\n",
    "**Redundancy** is the duplication of critical components or functions of a system with the intention of increasing the reliability of the system, usually in the form of a backup or fail-safe, or to improve actual system performance. For example, if there is only one copy of a file stored on a single server, then losing that server means losing the file. Since losing data is seldom a good thing, we can create duplicate or redundant copies of the file to solve this problem.\n",
    "\n",
    "Redundancy plays a key role in removing the single points of failure in the system and provides backups if needed in a crisis. For example, if we have two instances of a service running in production and one fails, the system can failover to the other one.\n",
    "\n",
    "<img src='imgs/rr.png' alt='rr' width=300 height=300>\n",
    "\n",
    "\n",
    "### Replication\n",
    "Replication means sharing information to ensure consistency between redundant resources, such as software or hardware components, to improve reliability, fault-tolerance, or accessibility.\n",
    "\n",
    "The act of **duplicating the data** from one database server to others. This is sometimes used to **increase the redundancy** of our system and tolerate regional failures for instance. Other times, we can use replication to move data closer to our clients, thus decreasing the latency of accessing specific data.\n",
    "\n",
    "Replication is widely used in many database management systems (DBMS), usually with a primary-replica relationship between the original and the copies. The primary server gets all the updates, which then ripple through to the replica servers. Each replica outputs a message stating that it has received the update successfully, thus allowing the sending of subsequent updates.\n",
    "\n",
    "### Sharding \n",
    "Sometimes called **data partitioning**, sharding is the act of splitting a database into two or more pieces called **shards** and is typically done to **increase the throughput** of our database. Popular sharding strategies include:\n",
    "\n",
    "* Sharding based on a client's region\n",
    "* Sharding based on the type of data being stored (e.g: user data gets stored in one shard, payments data gets stored in another shard)\n",
    "* Sharding based on the hash of a column (only for structured data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Proxies"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "A **proxy server** is an intermediate server between the client and the back-end server. Clients connect to proxy servers to make a request for a service like a web page, file, connection, etc. In short, a proxy server is a piece of software or hardware that acts as an intermediary for requests from clients seeking resources from other servers.\n",
    "\n",
    "Typically, proxies are used to filter requests, log requests, or sometimes transform requests (by adding/removing headers, encrypting/decrypting, or compressing a resource). Another advantage of a proxy server is that its cache can serve a lot of requests. If multiple clients access a particular resource, the proxy server can cache it and serve it to all the clients without going to the remote server.\n",
    "\n",
    "Proxies can reside on the client’s local server or anywhere between the client and the remote servers. Here are a few famous types of proxy servers:\n",
    "\n",
    "### Forward Proxy\n",
    "A server that sits between a client and servers and acts on behalf of the client, typically used to mask the client's identity (IP address). Note that forward proxies are often referred to as just proxies.\n",
    "\n",
    "### Open Proxy \n",
    "An **open proxy** is a proxy server that is accessible by any Internet user. Generally, a proxy server only allows users within a network group (i.e. a closed proxy) to store and forward Internet services such as DNS or web pages to reduce and control the bandwidth used by the group. With an open proxy, however, any user on the Internet is able to use this forwarding service. There two famous open proxy types:\n",
    "- **Anonymous Proxy** - Thіs proxy reveаls іts іdentіty аs а server but does not dіsclose the іnіtіаl IP аddress. Though thіs proxy server cаn be dіscovered eаsіly іt cаn be benefіcіаl for some users аs іt hіdes their IP аddress.\n",
    "- **Trаnspаrent Proxy** – Thіs proxy server аgаіn іdentіfіes іtself, аnd wіth the support of HTTP heаders, the fіrst IP аddress cаn be vіewed. The mаіn benefіt of usіng thіs sort of server іs іts аbіlіty to cаche the websіtes.\n",
    "\n",
    "### Reverse Proxy\n",
    "A server that sits between clients and servers and acts on behalf of the servers, typically used for logging, load balancing, or caching.\n",
    "\n",
    "A **reverse proxy** (or **surrogate**) is a proxy server that appears to clients to be an ordinary server. Reverse proxies forward requests to one or more ordinary servers that handle the request. The response from the proxy server is returned as if it came directly from the original server, leaving the client with no knowledge of the original server.retrieves resources on behalf of a client from one or more servers. These resources are then returned to the client, appearing as if they originated from the proxy server itself.\n",
    "\n",
    "### Nginx\n",
    "\n",
    "Pronounced \"engine X\", **Nginx** is a very popular webserver that's often used as a **reverse proxy** and **load balancer**."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Load Balancers"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### Load Balancer\n",
    "\n",
    "A type of **reverse proxy** that distributes traffic across servers. Load balancers can be found in many parts of a system, from the DNS layer all the way to the database layer.\n",
    "\n",
    "**Load Balancer (LB)** is another critical component of any distributed system. It helps to spread the traffic across a cluster of servers to improve responsiveness and availability of applications, websites or databases. LB also keeps track of the status of all the resources while distributing requests. If a server is not available to take new requests or is not responding or has elevated error rate, LB will stop sending traffic to such a server.\n",
    "\n",
    "Typically a load balancer sits between the client and the server accepting incoming network and application traffic and distributing the traffic across multiple backend servers using various algorithms. By balancing application requests across multiple servers, a load balancer reduces individual server load and prevents any one application server from becoming a single point of failure, thus improving overall application availability and responsiveness.\n",
    "\n",
    "<img src='imgs/lb1.png' alt='lb1' width=600 height=400>\n",
    "\n",
    "To utilize full scalability and redundancy, we can try to balance the load at each layer of the system. We can add LBs at three places:\n",
    "- Between the user and the web server\n",
    "- Between web servers and an internal platform layer, like application servers or cache servers\n",
    "- Between internal platform layer and database.\n",
    "\n",
    "<img src='imgs/lb2.png' alt='lb2' width=700 height=400>\n",
    "\n",
    "### Benefits of Load Balancing\n",
    "* Users experience faster, uninterrupted service. Users won’t have to wait for a single struggling server to finish its previous tasks. Instead, their requests are immediately passed on to a more readily available resource.\n",
    "\n",
    "\n",
    "* Service providers experience less downtime and higher throughput. Even a full server failure won’t affect the end user experience as the load balancer will simply route around it to a healthy server.\n",
    "\n",
    "\n",
    "* Load balancing makes it easier for system administrators to handle incoming requests while decreasing wait time for users.\n",
    "\n",
    "\n",
    "* Smart load balancers provide benefits like predictive analytics that determine traffic bottlenecks before they happen. As a result, the smart load balancer gives an organization actionable insights. These are key to automation and can help drive business decisions.\n",
    "\n",
    "\n",
    "* System administrators experience fewer failed or stressed components. Instead of a single device performing a lot of work, load balancing has several devices perform a little bit of work.\n",
    "\n",
    "### Application server cache\n",
    "Placing a cache directly on a request layer node enables the local storage of response data. Each time a request is made to the service, the node will quickly return locally cached data if it exists. If it is not in the cache, the requesting node will fetch the data from the disk. The cache on one request layer node could also be located both in memory (which is very fast) and on the node’s local disk (faster than going to network storage).\n",
    "\n",
    "What happens when we expand this to many nodes? If the request layer is expanded to multiple nodes, it’s still quite possible to have each node host its own cache. However, if our load balancer randomly distributes requests across the nodes, the same request will go to different nodes, thus increasing **cache misses**. Two choices for overcoming this hurdle are **global caches** and **distributed caches**."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Load Balancing Algorithms"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "#### How does the load balancer choose the backend server?\n",
    "Load balancers consider two factors before forwarding a request to a backend server. They will first ensure that the server they choose is actually responding appropriately to requests and then use a pre-configured algorithm to select one from the set of healthy servers. We will discuss these algorithms shortly.\n",
    "\n",
    "#### Health Checks\n",
    "Load balancers should only forward traffic to `healthy` backend servers. To monitor the health of a backend server, **health checks** regularly attempt to connect to backend servers to ensure that servers are listening. If a server fails a health check, it is automatically removed from the pool, and traffic will not be forwarded to it until it responds to the health checks again.\n",
    "\n",
    "### Server-Selection Strategy\n",
    "\n",
    "How a load balancer chooses servers when distributing traffic amongst multiple servers. Commonly used strategies include **round robin, random selection, performance-based selection** (choosing the server with the best performance metrics, like the fastest response time or the least amount of traffic), and **IP-based routing**.\n",
    "\n",
    "There is a variety of load balancing methods, which use different algorithms for different needs.\n",
    "\n",
    "* **Least Connection Method** — This method directs traffic to the server with the fewest active connections. This approach is quite useful when there are a large number of persistent client connections which are unevenly distributed between the servers.\n",
    "\n",
    "\n",
    "* **Least Response Time Method** — This algorithm directs traffic to the server with the fewest active connections and the lowest average response time.\n",
    "\n",
    "\n",
    "* **Least Bandwidth Method** - This method selects the server that is currently serving the least amount of traffic measured in megabits per second (Mbps).\n",
    "\n",
    "\n",
    "* **Round Robin Method** — This method cycles through a list of servers and sends each new request to the next server. When it reaches the end of the list, it starts over at the beginning. It is most useful when the servers are of equal specification and there are not many persistent connections.\n",
    "\n",
    "\n",
    "* **Weighted Round Robin Method** — The weighted round-robin scheduling is designed to better handle servers with different processing capacities. Each server is assigned a weight (an integer value that indicates the processing capacity). Servers with higher weights receive new connections before those with less weights and servers with higher weights get more connections than those with less weights.\n",
    "\n",
    "\n",
    "* **IP Hash** — Under this method, a hash of the IP address of the client is calculated to redirect the request to a server.\n",
    "\n",
    "### Hot Spot\n",
    "When distributing a workload across a set of servers, that workload might be spread unevenly. This can happen if our **sharding key** or our **hashing function** are suboptimal, or if our workload is naturally skewed: some servers will receive a lot more traffic than others, thus creating a **hot spot**.\n",
    "\n",
    "### Redundant Load Balancers\n",
    "The load balancer can be a single point of failure; to overcome this, a second load balancer can be connected to the first to form a cluster. Each LB monitors the health of the other and, since both of them are equally capable of serving traffic and failure detection, in the event the main load balancer fails, the second load balancer takes over."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Load Balancing and SSL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### SSL\n",
    "**Secure Sockets Layer (SSL)** is the standard security technology for establishing an encrypted link between a web server and a browser. SSL traffic is often decrypted at the load balancer. When a load balancer decrypts traffic before passing the request on, it is called **SSL termination**. The load balancer saves the web servers from having to expend the extra CPU cycles required for decryption. This improves application performance.\n",
    "\n",
    "However, SSL termination comes with a security concern. The traffic between the load balancers and the web servers is no longer encrypted. This can expose the application to possible attack. However, the risk is lessened when the load balancer is within the same data center as the web servers.\n",
    "\n",
    "Another solution is the **SSL pass-through**. The load balancer merely passes an encrypted request to the web server. Then the web server does the decryption. This uses more CPU power on the web server. But organizations that require extra security may find the extra overhead worthwhile.\n",
    "\n",
    "### Load Balancing and Security\n",
    "Load Balancing plays an important security role as computing moves evermore to the cloud. The off-loading function of a load balancer defends an organization against **distributed denial-of-service (DDoS) attacks**. It does this by shifting attack traffic from the corporate server to a public cloud provider. DDoS attacks represent a large portion of cybercrime as their number and size continues to rise. Hardware defense, such as a perimeter firewall, can be costly and require significant maintenance. Software load balancers with cloud offload provide efficient and cost-effective protection.\n",
    "\n",
    "### DNS Load Balancing vs Hardware Load Balancing\n",
    "**DNS load balancing** is a software-defined approach to load balancing where client requests to a domain within the **Domain Name System (DNS)** are distributed across different server machines. The DNS system sends a different version of the list of IP addresses each time it responds to a new client request using the **round-robin method**, therefore distributing the DNS requests evenly to different servers to handle the overall load. This in turn provides DNS load balancing failover protection through automatic removal of non-responsive servers.\n",
    "\n",
    "DNS load balancing differs from hardware load balancing in a few instances, although both can be a very effective solution for distributing traffic. One main advantage of DNS level load balancing is the scalability and price. A DNS load balancer distributes traffic to several different IP addresses, whereas the hardware solution uses a single IP address and splits traffic leading to it on multiple servers. As for pricing, hardware load balancers require a large upfront cost whereas DNS load balancers can be scaled as needed."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Key-Value Stores"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "Data is stored in an array of **key-value pairs**. The `key` is an attribute name which is linked to a `value`. A **Key-Value Store** is a flexible NoSQL database that's often used for caching and dynamic configuration. Popular options include DynamoDB, Etcd, Redis, and Zookeeper.\n",
    "\n",
    "### Etcd\n",
    "\n",
    "Etcd is a strongly consistent and highly available key-value store that's often used to implement leader election in a system.\n",
    "\n",
    "### Redis\n",
    "\n",
    "An in-memory key-value store. Does offer some persistent storage options but is typically used as a really fast, best-effort caching solution. Redis is also often used to implement **rate limiting**.\n",
    "\n",
    "### Zookeeper\n",
    "\n",
    "Zookeeper is a strongly consistent, highly available key-value store. It's often used to store important configuration or to perform leader election."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Specialized Storage Paradigms"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Blob Storage\n",
    "\n",
    "Widely used kind of storage, in small and large scale systems. They don't really count as databases per se, partially because they only allow the user to store and retrieve data based on the name of the **blob**. This is sort of like a key-value store but usually blob stores have different guarantees. They might be slower than KV stores but values can be megabytes large (or sometimes gigabytes large). Usually people use this to store things like **large binaries, database snapshots, or images** and other **static assets** that a website might have.\n",
    "\n",
    "Blob storage is rather complicated to have on premise, and only giant companies like Google and Amazon have infrastructure that supports it. For example, **GCS** or **S3**. These are blob storage services hosted by Google and Amazon respectively, that cost money depending on how much storage we use and how often we store and retrieve blobs from that storage.\n",
    "\n",
    "### Time Series Database\n",
    "\n",
    "A **TSDB** is a special kind of database optimized for storing and analyzing time-indexed data: data points that specifically occur at a given moment in time. Examples of TSDBS are **InfluxDB**, **Prometheus**, and **Graphite**.\n",
    "\n",
    "### Document Databases\n",
    "In these databases, data is stored in documents (instead of rows and columns in a table) and these documents are grouped together in collections. Each document can have an entirely different structure. Document databases include the **CouchDB** and **MongoDB**.\n",
    "\n",
    "### Wide-Column Databases\n",
    "Instead of ‘tables,’ in columnar databases we have column families, which are containers for rows. Unlike relational databases, we don’t need to know all the columns up front and each row doesn’t have to have the same number of columns. Columnar databases are best suited for analyzing large datasets - big names include **Cassandra** and **HBase**.\n",
    "\n",
    "### Graph Database\n",
    "\n",
    "A type of database that stores data following the graph data model. Data entries in a graph database can have explicitly defined **relationships**, much like nodes in a graph can have edges. Data is saved in graph structures with **nodes** (entities), **properties** (information about the entities), and **lines** (connections between the entities). Examples of graph database include **Neo4J** and **InfiniteGraph**.\n",
    "\n",
    "Graph databases take advantage of their underlying graph structure to perform complex queries on deeply connected data very fast.\n",
    "\n",
    "Graph databases are thus often preferred to relational databases when dealing with systems where data points naturally form a graph and have multiple levels of relationships. For example, social networks. \n",
    "\n",
    "### Cypher\n",
    "A **graph query language** that was originally developed for the **Neo4j graph database**, but that has since been standardized to be used with other graph databases in an effort to make it the **SQL for graphs.**\n",
    "\n",
    "Cypher queries are often much simpler than their SQL counterparts. Example Cypher query to find data in Neo4j, a popular graph database:\n",
    "\n",
    "> MATCH (some_node:SomeLabel)-[:SOME RELATIONSHIP]->(some other_node: SomeLabel (some property: 'value' })\n",
    "\n",
    "### Spatial Database\n",
    "\n",
    "A type of database optimized for storing and querying spatial data like locations on a map. Spatial databases rely on spatial indexes like **quadtrees** to quickly perform spatial queries like finding all locations in the vicinity of a region.\n",
    "\n",
    "### Quadtree\n",
    "\n",
    "A tree data structure most commonly used to index two-dimensional spatial data. Each node in a quadtree has either zero children nodes and is therefore a leaf node) or exactly four children nodes.\n",
    "\n",
    "Typically, quadtree nodes contain some form of spatial data-for example, locations on a map with a maximum capacity of some specified number **n**. So long as nodes aren't at capacity, they remain leaf nodes; once they reach capacity, they're given four children nodes, and their data entries are split across the four children nodes.\n",
    "\n",
    "A quadtree lends itself well to storing spatial data because it can be represented as a **grid** filled with rectangles that are recursively subdivided into four sub-rectangles, where each quadtree node is represented by a **rectangle** and each rectangle represents a **spatial region**. Assuming we're storing locations in the world, we can imagine a quadtree with a maximum node-capacity n as follows:\n",
    "\n",
    "* The **root node**, which represents the entire world, is the outermost rectangle.\n",
    "* If the entire world has more than n locations, the outermost rectangle is divided into four quadrants, each representing a region of the world\n",
    "* So long as a region has more than n locations, its corresponding rectangle is subdivided into four quadrants (the corresponding node in the quadtree is given four **children nodes**).\n",
    "* Regions that have fewer than n locations are undivided rectangles (**leaf nodes**).\n",
    "* The parts of the grid that have many subdivided rectangles represent densely populated areas (like cities), while the parts of the grid that have few subdivided rectangles represent sparsely populated areas (like rural areas).\n",
    "\n",
    "Finding a given location in a perfect quadtree is an extremely fast operation that runs in **$log_{4}(x)$** time (where **x** is the total number of locations), since quadtree nodes have four children nodes.\n",
    "\n",
    "### S3\n",
    "S3 is a blob storage service provided by Amazon through Amazon Web Services (AWS). Amazon S3 has a simple web services interface that we can use to store and retrieve any amount of data, at any time, from anywhere on the web. It gives any developer access to the same highly scalable, reliable, fast, inexpensive data storage infrastructure that Amazon uses to run its own global network of web sites.\n",
    "\n",
    "### Google Cloud Storage\n",
    "\n",
    "GCS is a blob storage service provided by Google.\n",
    "\n",
    "### InfluxDB\n",
    "\n",
    "A popular open-source time series database.\n",
    "\n",
    "### Prometheus\n",
    "\n",
    "A popular open-source time series database, typically used for monitoring purposes.\n",
    "\n",
    "### Neo4j\n",
    "\n",
    "A popular graph database that consists of nodes, relationships, properties, and labels.\n",
    "\n",
    "### MongoDB\n",
    "\n",
    "A NoSQL database with powerful querying through a JavaScript-like language. Consistency guarantees depend on the settings that the database is setup with."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Distributed Systems"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "A **distributed system** is a system whose components are located on different networked computers, which communicate and coordinate their actions by passing messages to one another from any system. The components interact with one another in order to achieve a common goal. Key characteristics of a distributed system include Scalability, Reliability, Availability, Efficiency, and Manageability.\n",
    "\n",
    "<img src='imgs/distributed_system.png' alt='vshs' width=500 height=500>\n",
    "\n",
    "### Scalability\n",
    "**Scalability** is the capability of a system, process, or a network to grow and manage increased demand. Any distributed system that can continuously evolve in order to support the growing amount of work is considered to be scalable.\n",
    "\n",
    "A system may have to scale because of many reasons like increased data volume or increased amount of work, e.g., number of transactions. A scalable system would like to achieve this scaling without performance loss.\n",
    "\n",
    "Generally, the performance of a system, although designed (or claimed) to be scalable, declines with the system size due to the management or environment cost. For instance, network speed may become slower because machines tend to be far apart from one another. More generally, some tasks may not be distributed, either because of their inherent atomic nature or because of some flaw in the system design. At some point, such tasks would limit the speed-up obtained by distribution. A scalable architecture avoids this situation and attempts to balance the load on all the participating nodes evenly.\n",
    "\n",
    "### Horizontal vs. Vertical Scaling\n",
    "**Horizontal scaling** means that we scale by adding more servers into our pool of resources whereas **Vertical scaling** means that we scale by adding more power (CPU, RAM, Storage, etc.) to an existing server.\n",
    "\n",
    "With horizontal-scaling it is often easier to scale dynamically by adding more machines into the existing pool; Vertical-scaling is usually limited to the capacity of a single server and scaling beyond that capacity often involves downtime and comes with an upper limit.\n",
    "\n",
    "Good examples of horizontal scaling are **Cassandra** and **MongoDB** as they both provide an easy way to scale horizontally by adding more machines to meet growing needs. Similarly, a good example of vertical scaling is **MySQL** as it allows for an easy way to scale vertically by switching from smaller to bigger machines. However, this process often involves downtime.\n",
    "\n",
    "<img src='imgs/vshs.png' alt='vshs' width=400 height=400>\n",
    "\n",
    "### Reliability\n",
    "**Reliability** is the probability a system will fail in a given period. In simple terms, a distributed system is considered reliable if it keeps delivering its services even when one or several of its software or hardware components fail. Reliability represents one of the main characteristics of any distributed system, since in such systems any failing machine can always be replaced by another healthy one, ensuring the completion of the requested task.\n",
    "\n",
    "Take the example of a large electronic commerce store (like Amazon), where one of the primary requirement is that any user transaction should never be canceled due to a failure of the machine that is running that transaction. For instance, if a user has added an item to their shopping cart, the system is expected not to lose it. A reliable distributed system achieves this through redundancy of both the software components and data. If the server carrying the user’s shopping cart fails, another server that has the exact replica of the shopping cart should replace it.\n",
    "\n",
    "Obviously, redundancy has a cost and a reliable system has to pay that to achieve such resilience for services by eliminating every single point of failure.\n",
    "\n",
    "\n",
    "### Availability\n",
    "**Availability** is the time a system remains operational to perform its required function in a specific period. It is a simple measure of the percentage of time that a system, service, or a machine remains operational under normal conditions. An aircraft that can be flown for many hours a month without much downtime can be said to have a high availability. Availability takes into account maintainability, repair time, spares availability, and other logistics considerations. If an aircraft is down for maintenance, it is considered not available during that time.\n",
    "\n",
    "Reliability is availability over time considering the full range of possible real-world conditions that can occur. An aircraft that can make it through any possible weather safely is more reliable than one that has vulnerabilities to possible conditions.\n",
    "\n",
    "### Reliability Vs. Availability\n",
    "If a system is reliable, it is available. However, if it is available, it is not necessarily reliable. In other words, high reliability contributes to high availability, but it is possible to achieve a high availability even with an unreliable product by minimizing repair time and ensuring that spares are always available when they are needed. \n",
    "\n",
    "Let’s take the example of an online retail store that has 99.99% availability for the first two years after its launch. However, the system was launched without any information security testing. The customers are happy with the system, but they don’t realize that it isn’t very reliable as it is vulnerable to likely risks. In the third year, the system experiences a series of information security incidents that suddenly result in extremely low availability for extended periods of time. This results in reputational and financial damage to the customers.\n",
    "\n",
    "\n",
    "### Efficiency \n",
    "To understand how to measure the **efficiency** of a distributed system, let’s assume we have an operation that runs in a distributed manner and delivers a set of items as result. Two standard measures of its efficiency are the **response time** (or **latency**) that denotes the delay to obtain the first item and the **throughput** (or **bandwidth**) which denotes the number of items delivered in a given time unit (e.g., a second). The two measures correspond to the following unit costs:\n",
    "- Number of messages globally sent by the nodes of the system regardless of the message size.\n",
    "- Size of messages representing the volume of data exchanges.\n",
    "\n",
    "The complexity of operations supported by distributed data structures (e.g., searching for a specific key in a distributed index) can be characterized as a function of one of these cost units. Generally speaking, the analysis of a distributed structure in terms of ‘number of messages’ is over-simplistic. It ignores the impact of many aspects, including the network topology, the network load, and its variation, the possible heterogeneity of the software and hardware components involved in data processing and routing, etc. However, it is quite difficult to develop a precise cost model that would accurately take into account all these performance factors; therefore, we have to live with rough but robust estimates of the system behavior.\n",
    "\n",
    "### Serviceability or Manageability\n",
    "Another important consideration while designing a distributed system is how easy it is to operate and maintain. **Serviceability** or **manageability** is the simplicity and speed with which a system can be repaired or maintained; if the time to fix a failed system increases, then availability will decrease. Things to consider for manageability are the ease of diagnosing and understanding problems when they occur, ease of making updates or modifications, and how simple the system is to operate (i.e., does it routinely operate without failure or exceptions?).\n",
    "\n",
    "Early detection of faults can decrease or avoid system downtime. For example, some enterprise systems can automatically call a service center (without human intervention) when the system experiences a system fault.\n",
    "\n",
    "### Virtual Machine\n",
    "\n",
    "A **VM** is a form of computer inside of a computer. It is a program that we run on a machine that completely emulates a new **kernel** and **operating system**. Very useful when isolating programs from one another while having them share the same physical machine.\n",
    "\n",
    "### Worker Pool Pattern\n",
    "\n",
    "Similar to the **Task Queue Pattern**. In this design, a pool of **workers**, usually themselves servers, take tasks off of a single shared queue and process those tasks independently. In order to ensure that every task gets done at least once despite potential partitions between queue and workers, the workers must confirm the status of the task after it is done (usually **success** or **failure**)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## CAP theorem"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "In **distributed systems**, different types of failures can occur, e.g., servers can crash or fail permanently, disks can go bad resulting in data losses, or network connection can be lost, making a part of the system inaccessible. How can a distributed system model itself to get the maximum benefits out of different resources available?\n",
    "\n",
    "**CAP theorem** states that it is impossible for a distributed system to simultaneously provide all three of the following desirable properties:\n",
    "\n",
    "* **Consistency ( C )**: All nodes see the same data at the same time. It is equivalent to having a single up-to-date copy of the data.\n",
    "\n",
    "* **Availability ( A )**: Every request received by a non-failing node in the system must result in a response. Even when severe network failures occur, every request must terminate.\n",
    "\n",
    "* **Partition tolerance ( P )**: A partition-tolerant system continues to operate despite partial system failure or arbitrary message loss. Such a system can sustain any network failure that does not result in a failure of the entire network. Data is sufficiently replicated across combinations of nodes and networks to keep the system up through intermittent outages.\n",
    "\n",
    "According to the CAP theorem, any distributed system needs to pick two out of the three properties. The three options are **CA**, **CP**, and **AP**. However, CA is not really a coherent option, as a system that is not partition-tolerant will be forced to give up either Consistency or Availability in the case of a network partition. Therefore, the theorem can really be stated as: In the presence of a network partition, a distributed system must choose either **Consistency** or **Availability**.\n",
    "\n",
    "One thing to keep in mind is that some levels of consistency are still achievable with high availability, but strong consistency is much harder.\n",
    "\n",
    "<img src='imgs/cap.png' alt='cap' width=300 height=500>\n",
    "\n",
    "We cannot build a general data store that is continually available, sequentially consistent, and tolerant to any partition failures. We can only build a system that has any two of these three properties. Because, to be consistent, all nodes should see the same set of updates in the same order. But if the network loses a partition, updates in one partition might not make it to the other partitions before a client reads from the out-of-date partition after having read from the up-to-date one. The only thing that can be done to cope with this possibility is to stop serving requests from the out-of-date partition, but then the service is no longer 100% available."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## PACELC Theorem"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "We cannot avoid **partition** in a distributed system, therefore, according to the CAP theorem, a distributed system should choose between consistency or availability. **ACID (Atomicity, Consistency, Isolation, Durability) databases chose consistency** (refuse response if it cannot check with peers), while **BASE (Basically Available, Soft-state, Eventually consistent) databases chose availability** (respond with local data without ensuring it is the latest with its peers). One place where the CAP theorem is silent is what happens when there is no network partition? What choices does a distributed system have when there is no partition?\n",
    "\n",
    "The **PACELC theorem** states that in a system that replicates data:\n",
    "* if there is a **partition** `P`, a distributed system can tradeoff between **availability** and **consistency** (i.e., `A` and `C`);\n",
    "* else `E`, when the system is running normally in the absence of partitions, the system can tradeoff between **latency** `L` and **consistency** `C`.\n",
    "\n",
    "The first part of the theorem **PAC** is the same as the **CAP theorem**, and the **ELC** is the **extension**. The whole thesis is assuming we maintain high availability by replication. So, when there is a failure, CAP theorem prevails. But if not, we still have to consider the tradeoff between consistency and latency of a replicated system.\n",
    "\n",
    "Some Examples are:\n",
    "* **Dynamo** and **Cassandra** are **PA/EL** systems: They choose availability over consistency when a partition occurs; otherwise, they choose lower latency.\n",
    "* **BigTable** and **HBase** are **PC/EC** systems: They will always choose consistency, giving up availability and lower latency.\n",
    "* **MongoDB** is **PA/EC**: In case of a network partition, it chooses availability, but otherwise guarantees consistency."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Data partitioning"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "**Data partitioning** is a technique to break up a big database (DB) into many smaller parts. It is the process of splitting up a DB/table across multiple machines to improve the manageability, performance, availability, and load balancing of an application. The justification for data partitioning is that, after a certain scale point, it is cheaper and more feasible to scale horizontally by adding more machines than to grow it vertically by adding beefier servers.\n",
    "\n",
    "### Partitioning Methods\n",
    "There are many different schemes one could use to decide how to break up an application database into multiple smaller DBs. Below are three of the most popular schemes used by various large scale applications.\n",
    "\n",
    "**Horizontal partitioning**: In this scheme, we put different rows into different tables. For example, if we are storing different places in a table, we can decide that locations with ZIP codes less than 10000 are stored in one table and places with ZIP codes greater than 10000 are stored in a separate table. This is also called a **range based partitioning** as we are storing different ranges of data in separate tables. Horizontal partitioning is also called as **Data Sharding**.\n",
    "\n",
    "The key problem with this approach is that if the value whose range is used for partitioning isn’t chosen carefully, then the partitioning scheme will lead to unbalanced servers. In the previous example, splitting location based on their zip codes assumes that places will be evenly distributed across the different zip codes. This assumption is not valid as there will be a lot of places in a thickly populated area like Manhattan as compared to its suburb cities.\n",
    "\n",
    "**Vertical Partitioning**: In this scheme, we divide our data to store tables related to a specific feature in their own server. For example, if we are building Instagram like application - where we need to store data related to users, photos they upload, and people they follow - we can decide to place user profile information on one DB server, friend lists on another, and photos on a third server.\n",
    "\n",
    "Vertical partitioning is straightforward to implement and has a low impact on the application. The main problem with this approach is that if our application experiences additional growth, then it may be necessary to further partition a feature specific DB across various servers (e.g. it would not be possible for a single server to handle all the metadata queries for 10 billion photos by 140 million users).\n",
    "\n",
    "**Directory Based Partitioning**: A loosely coupled approach to work around issues mentioned in the above schemes is to create a lookup service which knows our current partitioning scheme and abstracts it away from the DB access code. So, to find out where a particular data entity resides, we query the directory server that holds the mapping between each tuple key to its DB server. This loosely coupled approach means we can perform tasks like adding servers to the DB pool or changing our partitioning scheme without having an impact on the application.\n",
    "\n",
    "### Partitioning Criteria\n",
    "**Key or Hash-based partitioning**: Under this scheme, we apply a hash function to some key attributes of the entity we are storing; that yields the partition number. For example, if we have `100 DB servers` and our `ID` is a numeric value that gets incremented by one each time a new record is inserted. In this example, the hash function could be `ID % 100`, which will give us the server number where we can store/read that record. This approach should ensure a uniform allocation of data among servers. The fundamental problem with this approach is that it effectively fixes the total number of DB servers, since adding new servers means changing the hash function which would require redistribution of data and downtime for the service. A workaround for this problem is to use **Consistent Hashing**.\n",
    "\n",
    "**List partitioning**: In this scheme, each partition is assigned a list of values, so whenever we want to insert a new record, we will see which partition contains our key and then store it there. For example, we can decide all users living in Iceland, Norway, Sweden, Finland, or Denmark will be stored in a partition for the Nordic countries.\n",
    "\n",
    "**Round-robin partitioning**: This is a very simple strategy that ensures uniform data distribution. With `n` partitions, the `i` tuple is assigned to partition (`i mod n`).\n",
    "\n",
    "**Composite partitioning**: Under this scheme, we combine any of the above partitioning schemes to devise a new scheme. For example, first applying a list partitioning scheme and then a hash based partitioning. Consistent hashing could be considered a composite of hash and list partitioning where the hash reduces the key space to a size that can be listed.\n",
    "\n",
    "### Common Problems of Data Partitioning\n",
    "On a partitioned database, there are certain extra constraints on the different operations that can be performed. Most of these constraints are due to the fact that operations across multiple tables or multiple rows in the same table will no longer run on the same server. Below are some of the constraints and additional complexities introduced by partitioning:\n",
    "\n",
    "**Joins and Denormalization**: Performing joins on a database which is running on one server is straightforward, but once a database is partitioned and spread across multiple machines it is often not feasible to perform joins that span database partitions. Such joins will not be performance efficient since data has to be compiled from multiple servers. A common workaround for this problem is to denormalize the database so that queries that previously required joins can be performed from a single table. Of course, the service now has to deal with all the perils of denormalization such as data inconsistency.\n",
    "\n",
    "**Referential integrity**: Performing a cross-partition query on a partitioned database is not feasible, similarly, trying to enforce data integrity constraints such as foreign keys in a partitioned database can be extremely difficult.\n",
    "\n",
    "Most of RDBMS do not support foreign keys constraints across databases on different database servers. Which means that applications that require referential integrity on partitioned databases often have to enforce it in application code. Often in such cases, applications have to run regular SQL jobs to clean up dangling references.\n",
    "\n",
    "**Rebalancing**: There could be many reasons we have to change our partitioning scheme:\n",
    "* The data distribution is not uniform, e.g., there are a lot of places for a particular ZIP code that cannot fit into one database partition.\n",
    "* There is a lot of load on a partition, e.g., there are too many requests being handled by the DB partition dedicated to user photos.\n",
    "In such cases, either we have to create more DB partitions or have to rebalance existing partitions, which means the partitioning scheme changed and all existing data moved to new locations. Doing this without incurring downtime is extremely difficult. Using a scheme like directory based partitioning does make rebalancing a more palatable experience at the cost of increasing the complexity of the system and creating a new single point of failure (i.e. the lookup service/database)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Consistent Hashing using Virtual Nodes in Distributed systems"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "A naive approach will use a suitable hash function to map the data key to a number. Then, find the server by applying modulo on this number and the total number of servers. For example:\n",
    "\n",
    "<img src='imgs/ch1.png' alt='ch1' width=400 height=500>\n",
    "\n",
    "The scheme described in the above diagram solves the problem of finding a server for storing/retrieving the data. But when we add or remove a server, all our existing mappings will be broken. This is because the total number of servers will be changed, which was used to find the actual server storing the data. So to get things working again, we have to **remap all the keys** and move our data based on the new server count, which will be a **complete mess!**\n",
    "\n",
    "Distributed systems can use Consistent Hashing to distribute data across nodes. Consistent Hashing maps data to physical nodes and ensures that **only a small set of keys move when servers are added or removed**.\n",
    "\n",
    "Consistent Hashing stores the data managed by a distributed system in a ring. Each node in the ring is assigned a range of data. Here is an example of the **consistent hash ring**:\n",
    "\n",
    "<img src='imgs/ch2.png' alt='ch2' width=500 height=500>\n",
    "\n",
    "With consistent hashing, the ring is divided into smaller, predefined ranges. Each node is assigned one of these ranges. The start of the range is called a **token**. This means that each node will be assigned one token. The range assigned to each node is computed as follows:\n",
    "\n",
    "* **Range start**:  Token value\n",
    "* **Range end**:    Next token value - 1\n",
    "\n",
    "Here are the tokens and data ranges of the four nodes described in the above diagram:\n",
    "\n",
    "<img src='imgs/ch3.png' alt='ch3' width=300 height=500>\n",
    "\n",
    "Whenever the system needs to read or write data, the first step it performs is to apply the **MD5 hashing algorithm** to the key. The output of this hashing algorithm determines within which range the data lies and hence, on which node the data will be stored. As we saw above, each node is supposed to store data for a fixed range. Thus, the hash generated from the key tells us the node where the data will be stored.\n",
    "\n",
    "<img src='imgs/ch4.png' alt='ch4' width=900 height=500>\n",
    "\n",
    "The Consistent Hashing scheme described above works great when a node is added or removed from the ring, as in these cases, since only the next node is affected. For example, when a node is removed, the next node becomes responsible for all of the keys stored on the outgoing node. However, this scheme can result in **non-uniform data and load distribution**. This problem can be solved with the help of **Virtual nodes**.\n",
    "\n",
    "### Virtual nodes\n",
    "Adding and removing nodes in any distributed system is quite common. Existing nodes can die and may need to be decommissioned. Similarly, new nodes may be added to an existing cluster to meet growing demands. To efficiently handle these scenarios, Consistent Hashing makes use of **virtual nodes** (or **Vnodes**).\n",
    "\n",
    "As we saw above, the basic Consistent Hashing algorithm assigns a single token (or a consecutive hash range) to each physical node. This was a static division of ranges that requires calculating tokens based on a given number of nodes. This scheme made adding or replacing a node an expensive operation, as, in this case, we would like to rebalance and distribute the data to all other nodes, resulting in moving a lot of data. Here are a few potential issues associated with a manual and fixed division of the ranges:\n",
    "* **Adding or removing nodes**: Adding or removing nodes will result in recomputing the tokens causing a significant administrative overhead for a large cluster.\n",
    "* **Hotspots**: Since each node is assigned one large range, if the data is not evenly distributed, some nodes can become hotspots.\n",
    "* **Node rebuilding**: Since each node’s data might be replicated (for fault-tolerance) on a fixed number of other nodes, when we need to rebuild a node, only its replica nodes can provide the data. This puts a lot of pressure on the replica nodes and can lead to service degradation.\n",
    "\n",
    "To handle these issues, Consistent Hashing introduces a new scheme of distributing the tokens to physical nodes. Instead of assigning a single token to a node, the hash range is divided into multiple smaller ranges, and each physical node is assigned several of these smaller ranges. Each of these subranges is considered a **Vnode**. With Vnodes, instead of a node being responsible for just one token, it is responsible for many tokens (or subranges).\n",
    "\n",
    "<img src='imgs/vn1.png' alt='vn1' width=600 height=500>\n",
    "\n",
    "Practically, Vnodes are randomly distributed across the cluster and are generally non-contiguous so that no two neighboring Vnodes are assigned to the same physical node or rack. Additionally, nodes do carry replicas of other nodes for **fault tolerance**. Also, since there can be heterogeneous machines in the clusters, some servers might hold more Vnodes than others. The figure below shows how physical nodes A, B, C, D, & E use Vnodes of the Consistent Hash ring. Each physical node is assigned a set of Vnodes and each Vnode is replicated once.\n",
    "\n",
    "<img src='imgs/vn2.png' alt='vn2' width=500 height=500>\n",
    "\n",
    "Vnodes gives the following advantages:\n",
    "* As Vnodes help spread the load more evenly across the physical nodes on the cluster by dividing the hash ranges into smaller subranges, this **speeds up the rebalancing process** after adding or removing nodes. When a new node is added, it receives many Vnodes from the existing nodes to maintain a balanced cluster. Similarly, when a node needs to be rebuilt, instead of getting data from a fixed number of replicas, many nodes participate in the rebuild process.\n",
    "* Vnodes make it easier to maintain a cluster containing heterogeneous machines. This means, with Vnodes, we can assign a high number of sub-ranges to a powerful server and a lower number of sub-ranges to a less powerful server.\n",
    "* In contrast to one big range, since Vnodes help assign smaller ranges to each physical node, this decreases the probability of hotspots.\n",
    "\n",
    "### Data replication using Consistent Hashing\n",
    "To ensure highly available and durability, Consistent Hashing replicates each data item on multiple `N` nodes in the system where the value `N` is equivalent to the **replication factor**.\n",
    "\n",
    "The replication factor is the number of nodes that will receive the copy of the same data. For example, a replication factor of two means there are two copies of each data item, where each copy is stored on a different node.\n",
    "\n",
    "Each key is assigned to a **coordinator node** (generally the first node that falls in the hash range), which first stores the data locally and then replicates it to `N−1` clockwise **successor nodes** on the ring. This results in each node owning the region on the ring between it and its `Nth` predecessor. In an **eventually consistent system**, this replication is done **asynchronously** (in the background).\n",
    "\n",
    "In eventually consistent systems, copies of data don’t always have to be identical as long as they are designed to eventually become consistent. **In distributed systems, eventual consistency is used to achieve high availability**.\n",
    "\n",
    "<img src='imgs/ch5.png' alt='ch5' width=500 height=500>\n",
    "\n",
    "Consistent Hashing helps with efficiently partitioning and replicating data; therefore, any distributed system that needs to scale up or down or wants to achieve high availability through data replication can utilize Consistent Hashing. A few such examples could be:\n",
    "* Any system working with a set of storage (or database) servers and needs to scale up or down based on the usage, e.g., the system could need more storage during Christmas because of high traffic.\n",
    "* Any distributed system that needs dynamic adjustment of its cache usage by adding or removing cache servers based on the traffic load.\n",
    "* Any system that wants to replicate its data shards to achieve high availability.\n",
    "\n",
    "**Amazon’s Dynamo** and **Apache Cassandra** use Consistent Hashing to distribute and replicate data across nodes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Quorum"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "In Distributed Systems, data is replicated across multiple servers for fault tolerance and high availability. Once a system decides to maintain multiple copies of data, another problem arises: how to make sure that all replicas are consistent, i.e., if they all have the latest copy of the data and that all clients see the same view of the data?\n",
    "\n",
    "In a distributed environment, a **quorum** is the minimum number of servers on which a distributed operation needs to be performed successfully before declaring the operation’s overall success.\n",
    "\n",
    "Suppose a database is replicated on five machines. In that case, quorum refers to the minimum number of machines that perform the same action (**commit** or **abort**) for a given transaction in order to decide the final operation for that transaction. So, in a set of 5 machines, three machines form the majority quorum, and if they agree, we will commit that operation. **Quorum enforces the consistency requirement needed for distributed operations**.\n",
    "\n",
    "In systems with multiple replicas, there is a possibility that the user reads inconsistent data. For example, when there are three replicas, `R1`, `R2`, and `R3` in a cluster, and a user writes value `v1` to replica `R1`. Then another user reads from replica `R2` or `R3` which are still behind `R1` and thus will not have the value `v1`, so the second user will not get the consistent state of data.\n",
    "\n",
    "**What value should we choose for a quorum?** More than half of the number of nodes in the cluster: **(N/2+1)** where `N` is the total number of nodes in the cluster, for example:\n",
    "* In a 5-node cluster, three nodes must be online to have a majority.\n",
    "* In a 4-node cluster, three nodes must be online to have a majority.\n",
    "* With 5-node, the system can afford two node failures, whereas, with 4-node, it can afford only one node failure. Because of this logic, **it is recommended to always have an odd number of total nodes in the cluster**.\n",
    "\n",
    "Quorum is achieved when nodes follow the below protocol: **R + W > N**, where:\n",
    "`N` = nodes in the quorum group\n",
    "`W` = minimum write nodes\n",
    "`R` = minimum read nodes\n",
    "\n",
    "If a distributed system follows R + W > N rule, then **every read will see at least one copy of the latest value written**. For example, a common configuration could be (N=3, W=2, R=2) to ensure strong consistency. Here are a couple of other examples:\n",
    "* (N=3, W=1, R=3): fast write, slow read, not very durable\n",
    "* (N=3, W=3, R=1): slow write, fast read, durable\n",
    "\n",
    "The following two things should be kept in mind before deciding **read/write quorum**:\n",
    "\n",
    "* **R = 1** and **W = N** ⇒ full replication (**write-all, read-one**): undesirable when servers can be unavailable because writes are not guaranteed to complete.\n",
    "* **Best performance** (throughput/availability) when **1 < r < w < n**, because reads are more frequent than writes in most applications."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Leader Election"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "Distributed systems keep multiple copies of data for fault tolerance and higher availability. A system can use **quorum** to ensure data consistency between replicas, i.e., all reads and writes are not considered successful until a majority of nodes participate in the operation. However, using quorum can lead to another problem, that is, lower availability; at any time, the system needs to ensure that at least a majority of replicas are up and available, otherwise the operation will fail. Quorum is also not sufficient, as in certain failure scenarios, the client can still see inconsistent data.\n",
    "\n",
    "The process by which nodes in a cluster (for instance, servers in a set of servers) elect a so-called **leader** amongst them, responsible for the primary operations of the service that these nodes support. When correctly implemented, **leader election** guarantees that all nodes in the cluster know which one is the leader at any given time and can elect a new leader if the leader dies for whatever reason.\n",
    "\n",
    "This allow only a single server (called **leader**) to be responsible for data replication and to coordinate work. This leader becomes responsible for data replication and can act as the central point for all coordination. The **followers** only accept writes from the leader and serve as a backup. In case the leader fails, one of the followers can become the leader. In some cases, the follower can serve read requests for load balancing.\n",
    "\n",
    "<img src='imgs/leader.png' alt='leader' width=500 height=500>\n",
    "\n",
    "### Consensus Algorithm\n",
    "A type of complex algorithms used to have multiple entities agree on a single data value, like who the `leader` is amongst a group of machines. Two popular consensus algorithms are **Paxos** and **Raft**.\n",
    "\n",
    "### Paxos & Raft\n",
    "Two consensus algorithms that, when implemented correctly, allow for the **synchronization** of certain operations, even in a distributed setting."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Peer-To-Peer Networks"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "A collection of machines referred to as **peers** that divide a workload between themselves to presumably complete the workload faster than would otherwise be possible. Peer-to-peer networks are often used in file-distribution systems.\n",
    "\n",
    "### Gossip Protocol\n",
    "\n",
    "When a set of machines talk to each other in a uncoordinated manner in a cluster to spread information through a system without requiring a central source of data."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Polling And Streaming"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### Polling\n",
    "The act of fetching a resource or piece of data regularly at an interval to make sure our data is not too stale.\n",
    "\n",
    "### Ajax Polling\n",
    "\n",
    "**Polling** is a standard technique used by the vast majority of AJAX applications. The basic idea is that the client repeatedly polls (or requests) a server for data. The client makes a request and waits for the server to respond with data. If no data is available, an empty response is returned.\n",
    "\n",
    "* The client opens a connection and requests data from the server using regular HTTP.\n",
    "* The requested webpage sends requests to the server at regular intervals (e.g., 0.5 seconds).\n",
    "* The server calculates the response and sends it back, just like regular HTTP traffic.\n",
    "* The client repeats the above three steps periodically to get updates from the server.\n",
    "\n",
    "The problem with Polling is that the client has to keep asking the server for any new data. As a result, a lot of responses are empty, creating HTTP overhead.\n",
    "\n",
    "<img src='imgs/ajax_polling.png' alt='ajax_polling' width=500 height=500>\n",
    "\n",
    "\n",
    "### HTTP Long-Polling\n",
    "This is a variation of the traditional polling technique that allows the server to push information to a client whenever the data is available. With **Long-Polling**, the client requests information from the server exactly as in **normal polling**, but with the expectation that the server may not respond immediately. That’s why this technique is sometimes referred to as a **Hanging GET**.\n",
    "\n",
    "* If the server does not have any data available for the client, instead of sending an empty response, the server holds the request and waits until some data becomes available.\n",
    "* Once the data becomes available, a full response is sent to the client. The client then immediately re-request information from the server so that the server will almost always have an available waiting request that it can use to deliver data in response to an event.\n",
    "\n",
    "The basic life cycle of an application using HTTP Long-Polling is as follows:\n",
    "\n",
    "* The client makes an initial request using regular HTTP and then waits for a response.\n",
    "* The server delays its response until an update is available or a timeout has occurred.\n",
    "* When an update is available, the server sends a full response to the client.\n",
    "* The client typically sends a new long-poll request, either immediately upon receiving a response or after a pause to allow an acceptable **latency period**.\n",
    "* Each Long-Poll request has a **timeout**. The client has to reconnect periodically after the connection is closed due to timeouts.\n",
    "\n",
    "<img src='imgs/long_polling.png' alt='long_polling' width=500 height=500>\n",
    "\n",
    "### WebSockets\n",
    "WebSocket provides **Full duplex** communication channels over a single TCP connection. It provides a persistent connection between a client and a server that both parties can use to start sending data at any time. The client establishes a WebSocket connection through a process known as the **WebSocket handshake**. If the process succeeds, then the server and client can exchange data in both directions at any time. The WebSocket protocol enables communication between a client and a server with **lower overheads**, facilitating real-time data transfer from and to the server. This is made possible by providing a standardized way for the server to send content to the browser without being asked by the client and allowing for messages to be passed back and forth while keeping the connection open. In this way, a two-way (**bi-directional**) ongoing conversation can take place between a client and a server.\n",
    "\n",
    "### Server-Sent Events (SSEs)\n",
    "Under **SSEs** the client establishes a persistent and long-term connection with the server. The server uses this connection to send data to a client. If the client wants to send data to the server, it would require the use of another technology/protocol to do so.\n",
    "\n",
    "* Client requests data from a server using regular HTTP.\n",
    "* The requested webpage opens a connection to the server.\n",
    "* The server sends the data to the client whenever there’s new information available.\n",
    "\n",
    "SSEs are best when we need real-time traffic from the server to the client or if the server is generating data in a loop and will be sending multiple events to the client.\n",
    "\n",
    "### Streaming\n",
    "In networking, it usually refers to the act of continuously getting a feed of information from a server by keeping an open connection between the two machines or processes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Configuration"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "A set of parameters or constants that are critical to a system. Configuration is typically written in **JSON** or **YAML** and can be either **static**, meaning that it's hard-coded in and shipped with our system's application code (like frontend code, for instance), or **dynamic**, meaning that it lives outside of our system's application code.\n",
    "\n",
    "### JSON\n",
    "\n",
    "A file format heavily used in APIs and configuration. Stands for **JavaScript Object Notation**. Example:\n",
    "```json\n",
    "{\n",
    "\"version\": 1.0,\n",
    "\"name\": \"System Design\"\n",
    "}\n",
    "```\n",
    "\n",
    "### YAML\n",
    "\n",
    "A file format mostly used in configuration. Example:\n",
    "\n",
    "```yaml\n",
    "version: 1.0\n",
    "name: System Design\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Rate Limiting"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "The act of limiting the number of requests sent to or from a system. **Rate limiting** is most often used to limit the number of incoming requests in order to prevent **DoS attacks** and can be enforced at the IP-address level, at the user-account level, or at the region level. Rate limiting can also be implemented in tiers; for instance, a type of network request could be limited to 1 per second, 5 per 10 seconds, and 10 per minute.\n",
    "\n",
    "### DoS Attack\n",
    "Short for **denial-of-service attack**, a DoS attack is an attack in which a malicious user tries to bring down or damage a system in order to render it unavailable to users. Much of the time, it consists of flooding it with traffic. Some DoS attacks are easily preventable with rate limiting, while others can be far trickier to defend against.\n",
    "\n",
    "### DDoS Attack\n",
    "Short for **distributed denial-of-service attack**, a DDoS attack is a DoS attack in which the traffic flooding the target system comes from many different sources (like thousands of machines), making it much harder to defend against."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Logging And Monitoring"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### Logging\n",
    "\n",
    "The act of collecting and storing logs--useful information about events in our system. Typically our programs will output log messages to its STDOUT or STDERR pipes, which will automatically get aggregated into a **centralized logging solution**.\n",
    "\n",
    "### Monitoring\n",
    "\n",
    "The process of having visibility into a system's key metrics, monitoring is typically implemented by collecting important events in a system and aggregating them in human-readable charts.\n",
    "\n",
    "### Alerting\n",
    "\n",
    "The process through which system administrators get notified critical system issues occur. Alerting can be set up by defining specific thresholds on monitoring charts, past which alerts are sent to a communication channel like Slack,"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Extract Transform Load (ETL) and Extract Load Transform (ELT)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### Data Pipeline\n",
    "A data pipeline is a series of steps in which data is processed.\n",
    "\n",
    "### Data Validation\n",
    "The process of ensuring that data is present, correct & meaningful. \n",
    "\n",
    "Ensuring the quality of our data through automated validation checks is a critical step in building data pipelines at any organization.\n",
    "\n",
    "### ETL\n",
    "ETL is normally a continuous, ongoing process with a well-defined workflow. ETL first extracts data from homogeneous or heterogeneous data sources. Then, data is cleansed, enriched, transformed, and stored either back in the lake or in a data warehouse.\n",
    "\n",
    "### ELT\n",
    "\"ELT (Extract, Load, Transform) is a variant of ETL wherein the extracted data is first loaded into the target system. Transformations are performed after the data is loaded into the data warehouse. ELT typically works well when the target system is powerful enough to handle transformations. Analytical databases like Amazon Redshift and Google BigQuery.\n",
    "\n",
    "### Redshift\n",
    "Amazon Redshift is a fully managed, petabyte-scale data warehouse service in the cloud. We can start with just a few hundred gigabytes of data and scale to a petabyte or more. The first step to create a data warehouse is to launch a set of nodes, called an Amazon Redshift cluster. After we provision our cluster, we can upload our data set and then perform data analysis queries. Regardless of the size of the data set, Amazon Redshift offers fast query performance using the same SQL-based tools and business intelligence applications.\n",
    "\n",
    "### BigQuery\n",
    "BigQuery is a big data warehouse solution provided by Google."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Publish/Subscribe Pattern"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "Often shortened as **Pub/Sub**, the Publish/Subscribe pattern is a popular messaging model that consists of publishers and subscribers. **Publishers** publish messages to special **topics** (sometimes called **channels**) without caring about or even knowing who will read those messages, and **subscribers** subscribe to topics and read messages coming through those topics.\n",
    "\n",
    "Pub/Sub systems often come with very powerful guarantees like at-least-once delivery, persistent storage, ordering of messages, and replayability of messages.\n",
    "\n",
    "### Idempotent Operation\n",
    "\n",
    "An operation that has the same ultimate outcome regardless of how many times it's performed. If an operation can be performed multiple times without changing its overall effect, it's **idempotent**. Operations performed through a Pub/Sub messaging system typically have to be idempotent, since Pub/Sub systems tend to allow the same messages to be consumed multiple times.\n",
    "\n",
    "For example, increasing an integer value in a database is not an idempotent operation, since repeating this operation will not have the same effect as if it had been performed only once. Conversely, setting a value to `COMPLETE` is an idempotent operation, since repeating this operation will always yield the same result: the value will be `COMPLETE`.\n",
    "\n",
    "### Apache Kafka\n",
    "\n",
    "A distributed messaging system, very useful when using the **streaming** paradigm as opposed to **polling**. \n",
    "\n",
    "Apache Kafka is an open-source stream-processing software platform developed by Linkedin. The project aims to provide a unified, high-throughput, low-latency platform for handling real-time data feeds. Its storage layer is essentially a massively scalable pub/sub message queue designed as a distributed transaction log, making it highly valuable for enterprise infrastructures to process streaming data.\n",
    "\n",
    "### Cloud Pub/Sub\n",
    "\n",
    "A highly-scalable Pub/Sub messaging service created by Google. Guarantees at-least-once delivery of messages and supports `rewinding` in order to reprocess messages."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## MapReduce"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### File System\n",
    "\n",
    "An abstraction over a storage medium that defines how to manage data. While there exist many different types of file systems, most follow a **hierarchical structure** that consists of directories and files, like the Unix file system's structure.\n",
    "\n",
    "### MapReduce\n",
    "\n",
    "A popular framework for processing very large datasets in a distributed setting efficiently, quickly, and in a fault-tolerant manner. A MapReduce job is comprised of 3 main steps:\n",
    "\n",
    "* the **Map** step, which runs a **map function** on the various chunks of the dataset and transforms these chunks into intermediate **key-value pairs**.\n",
    "* the **Shuffle** step, which reorganizes the intermediate key-value pairs such that pairs of the same key are routed to the same machine in the final step.\n",
    "* the **Reduce** step, which runs a **reduce function** on the newly shuffled key-value pairs and transforms them into more meaningful data.\n",
    "\n",
    "The canonical example of a MapReduce use case is counting the number of occurrences of words in a large text file.\n",
    "\n",
    "When dealing with a MapReduce library, engineers and/or systems administrators only need to worry about the map and reduce functions, as well as their inputs and outputs. All other concerns, including the parallelization of tasks and the fault-tolerance of the MapReduce job, are abstracted away and taken care of by the MapReduce implementation.\n",
    "\n",
    "### Distributed File System\n",
    "\n",
    "A Distributed File System is an abstraction over a (usually large) cluster of machines that allows them to act like one large file system. The two most popular implementations of a DFS are the **Google File System** (**GFS**) and the **Hadoop Distributed File System** (**HDFS**).\n",
    "\n",
    "Typically, DFSS take care of the classic availability and replication guarantees that can be tricky to obtain in a distributed-system setting. The overarching idea is that files are split into **chunks** of a certain size (4MB ar 64MB, for instance), and those chunks are **sharded** across a large cluster of machines, A **central control plane** is in charge of deciding where each chunk resides, routing reads to the right nodes, and handling communication between machines\n",
    "\n",
    "Different DFS implementations have slightly different APIs and semantics, but they achieve the same common goal: extremely large scale persistent storage.\n",
    "\n",
    "### Hadoop\n",
    "\n",
    "A popular, open-source framework that supports MapReduce jobs and many other kinds of data-processing pipelines. Its central component is **HDFS** (**Hadoop Distributed File System**), on top of which other technologies have been developed."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Security And HTTPS"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### Man-In-The-Middle Attack\n",
    "\n",
    "An attack in which the attacker intercepts a line of communication that is thought to be private by its two communicating parties. If a malicious actor intercepted and mutated an IP packet on its way from a client to a server, that would be a man-in-the-middle attack.\n",
    "\n",
    "**MITM attacks** are the primary threat that encryption and HTTPS aim to defend against.\n",
    "\n",
    "### Symmetric Encryption\n",
    "\n",
    "A type of encryption that relies on only a **single key** to both encrypt and decrypt data. The key must be known to all parties involved in communication and must therefore typically be shared between the parties at one point or another.\n",
    "\n",
    "**Symmetric-key algorithms** tend to be faster than their asymmetric counterparts.\n",
    "\n",
    "The most widely used symmetric-key algorithms are part of the **Advanced Encryption Standard** (**AES**).\n",
    "\n",
    "### Asymmetric Encryption\n",
    "\n",
    "Also known as **public-key encryption**, asymmetric encryption relies on **two keys** -a **public key** and a **private key** -to encrypt and decrypt data. The keys are generated using **cryptographic algorithms** and are mathematically connected such that data encrypted with the public key can only be decrypted with the private key.\n",
    "\n",
    "While the private key must be kept secure to maintain the fidelity of this encryption paradigm, the public key can be openly shared. Asymmetric-key algorithms tend to be slower than their symmetric counterparts:\n",
    "\n",
    "### AES\n",
    "\n",
    "Stands for **Advanced Encryption Standard**, **AES** is a widely used encryption standard that has three symmetric-key algorithms (AES-128, AES-192, and AES-256).\n",
    "\n",
    "Of note, AES is considered to be the `gold standard` in encryption and is even used by the U.S. National Security Agency to encrypt top secret information.\n",
    "\n",
    "### HTTPS\n",
    "\n",
    "The **HyperText Transfer Protocol Secure** is an extension of **HTTP** that's used for secure communication online. It requires servers to have **trusted certificates** (usually **SSL certificates**) and uses the **Transport Layer Security** (**TLS**), a security protocol built on top of **TCP**, to encrypt data communicated between a client and a server.\n",
    "\n",
    "### TLS\n",
    "\n",
    "The **Transport Layer Security** is a security protocol over which HTTP runs in order to achieve secure communication online. `HTTP over TLS` is also known as **HTTPS**.\n",
    "\n",
    "### SSL Certificate\n",
    "\n",
    "A digital certificate granted to a server by a **certificate authority**. Contains the server's public key, to be used as part of the **TLS handshake** process in an **HTTPS** connection.\n",
    "\n",
    "An SSL certificate effectively confirms that a public key belongs to the server claiming it belongs to them. SSL certificates are a crucial defense against **man-in-the-middle attacks**.\n",
    "\n",
    "### Certificate Authority\n",
    "\n",
    "A trusted entity that signs digital certificates- namely, SSL certificates that are relied on in **HTTPS** connections.\n",
    "\n",
    "### TLS Handshake\n",
    "\n",
    "The process through which a client and a server communicating over **HTTPS** exchange encryption-related information and establish a secure communication. The typical steps in a TLS handshake are roughly as follows:\n",
    "\n",
    "* The client sends a **client hello** -a string of random bytes-to the server.\n",
    "\n",
    "* The server responds with a **server hello** -another string of random bytes-as well as its **SSL certificate**, which contains its **public key**.\n",
    "\n",
    "* The client verifies that the certificate was issued by a **certificate authority** and sends a **premaster secret** -yet another string of random bytes, this time encrypted with the server's public key-to the server.\n",
    "\n",
    "* The client and the server use the client hello, the server hello, and the premaster secret to then generate the **same symmetric encryption** session keys, to be used to encrypt and decrypt all data communicated during the remainder of the connection."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## API and SDK"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "Consider an analogy, SDK represents the entire house: all of the rooms, furniture, telephone lines, and other components. An API represents just the telephone lines that allow communication in and out of the house. A house needs telephone lines to communicate in and out and one house can have multiple telephone lines. A telephone line doesn’t need a house. \n",
    "\n",
    "### API\n",
    "**Application Program Interface** (**API**) is the engine under the hood that allows us to interact with external services using simple commands. API allows us to add specific functionalities to our application. In simple language, API is the messenger that takes requests, tells a system what we want to do, and then returns the response back to us. API’s help software engineers by preventing us from reinventing the wheel.\n",
    "\n",
    "### SDK\n",
    "**Software Development Kits** (**SDK**) is a set of tools, guidelines, and programs used to develop applications for a specific program. SDK is basically like a toolbox that calls an API for us. SDK’s can use one or many APIs, libraries, and other utilities. Companies make SDK’s available to developers in order for the developer to integrate with their services much easier. In some situations, it’s critical to use an SDK. For instance, if we want to develop an iOS application, we need the iOS SDK.\n",
    "\n",
    "\n",
    "\n",
    "<div class=\"alert alert-info\">\n",
    "In coding language, if an <b>SDK</b> is used for the application, it includes an <b>API</b> but if API is used for communication, it doesn’t include SDK. <b>SDK</b> is a kit that includes instructions that allow developers to create systems and <b>API</b> is purpose build for express use.\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## API Design"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### ACL\n",
    "\n",
    "Short for **Access-Control List**. This term is often used to refer to a permissioning model: which users in a system can perform which operations. For instance, APIs often come with ACLs defining which users can delete, edit, or view certain entities.\n",
    "\n",
    "### Pagination\n",
    "\n",
    "When a network request potentially warrants a really large response, the relevant API might be designed to return only a **single page** of that response (i.e., a limited portion of the response), accompanied by an identifier or token for the client to request the next page if desired.\n",
    "\n",
    "Pagination is often used when designing **List** endpoints. For instance, an endpoint to list videos on the YouTube Trending page could return a huge list of videos. This wouldn't perform very well on mobile devices due to the lower network speeds and simply wouldn't be optimal, since most users will only ever scroll through the first ten or twenty videos. So, the API could be designed to respond with only the first few videos of that list; in this case, we would say that the API response is **paginated**.\n",
    "\n",
    "### CRUD Operations\n",
    "\n",
    "Stands for **Create, Read, Update, Delete Operations**. These four operations often serve as the bedrock of a functioning system and therefore find themselves at the core of many APIs."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Microservice and Monolith Architecture"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### Microservice Architecture\n",
    "\n",
    "When a system is made up of many small web services that can be compiled and deployed independently. This is usually thought of as a counterpart of **monoliths**.\n",
    "\n",
    "### Monolith Architecture\n",
    "\n",
    "When a system is primarily made up of a single large web application that is compiled and rolled out as a unit. Typically a counterpart of **microservices**. Companies sometimes try to split up this monolith into microservices once it reaches a very large size in an attempt to increase **developer productivity**."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Bloom Filters"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "If we have a large set of **structured data** (identified by record IDs) stored in a set of data files, what is the most efficient way to know which file might contain our required data? We don’t want to read each file, as that would be slow, and we have to read a lot of data from the disk. One solution can be to build an index on each data file and store it in a separate index file. This index can map each record ID to its offset in the data file. Each index file will be sorted on the record ID. Now, if we want to search an ID in this index, the best we can do is a **Binary Search**. \n",
    "\n",
    "We can do better than this by using **Bloom filters** to quickly find if an element might be present in a **set**.\n",
    "\n",
    "The Bloom filter data structure tells whether an element may be in a set, or definitely is not. The only possible errors are **false positives**, i.e., a search for a nonexistent element might give an incorrect answer. With more elements in the filter, the error rate increases. An empty Bloom filter is a bit-array of `m bits`, all set to `0`. There are also `k` different hash functions, each of which maps a set element to one of the `m` bit positions.\n",
    "\n",
    "* To add an element, feed it to the hash functions to get `k` bit positions, and set the bits at these positions to `1`.\n",
    "* To test if an element is in the set, feed it to the hash functions to get `k` bit positions.\n",
    "    * If any of the bits at these positions is `0`, the element is definitely not in the set.\n",
    "    * If all are `1`, then the element may be in the set.\n",
    "    \n",
    "Here is a Bloom filter with three elements P, Q, and R. It consists of 20 bits and uses three hash functions. The colored arrows point to the bits that the elements of the set are mapped to.\n",
    "\n",
    "<img src='imgs/bloom.png' alt='bloom' width=500 height=500>\n",
    "\n",
    "* The element X definitely is not in the set, since it hashes to a bit position containing `0`.\n",
    "* For a fixed error rate, adding a new element and testing for membership are both **constant time operations**, and a filter with room for `n` elements requires **O(n) space**."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Heartbeat"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "In a distributed environment, work/data is distributed among servers. To efficiently route requests in such a setup, servers need to know what other servers are part of the system. Furthermore, servers should know if other servers are alive and working. In a decentralized system, whenever a request arrives at a server, the server should have enough information to decide which server is responsible for entertaining that request. This makes the timely detection of server failure an important task, which also enables the system to take corrective actions and move the data/work to another healthy server and stop the environment from further deterioration.\n",
    "\n",
    "Each server periodically sends a **heartbeat message** to a **central monitoring server** or other servers in the system to show that it is still alive and functioning.\n",
    "\n",
    "Heartbeating is one of the mechanisms for **detecting failures in a distributed system**. If there is a central server, all servers periodically send a heartbeat message to it. If there is no central server, all servers randomly choose a set of servers and send them a heartbeat message every few seconds. This way, if no heartbeat message is received from a server for a while, the system can suspect that the server might have crashed. If there is no heartbeat within a configured timeout period, the system can conclude that the server is not alive anymore and stop sending requests to it and start working on its replacement."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Checksum"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "In a distributed system, while moving data between components, it is possible that the data fetched from a node may arrive corrupted. This corruption can occur because of faults in a storage device, network, software, etc. \n",
    "\n",
    "A distributed system can ensure **data integrity** by calculating a **checksum** and store it with data. This makes sure that the client receives an error instead of corrupt data.\n",
    "\n",
    "To calculate a checksum, a cryptographic hash function like **MD5**, **SHA-1**, **SHA-256**, or **SHA-512** is used. The hash function takes the input data and produces a string (containing letters and numbers) of fixed length; this string is called the checksum.\n",
    "\n",
    "When a system is storing some data, it computes a checksum of the data and stores the checksum with the data. When a client retrieves data, it verifies that the data it received from the server matches the checksum stored. If not, then the client can opt to retrieve that data from another replica."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## File compression and Archiving"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "### File Compression\n",
    "**File compression** is a way to reduce the size of a file, or set of files. The advantages of compressed files are that they require less storage space and in the case of files that will be downloaded or uploaded, their reduced size reduces transmission time. There are two types of compression, **lossy** and **lossless**.\n",
    "\n",
    "#### Lossy Compression\n",
    "With **lossy compression**, some of the original data is discarded and the file cannot be reconstructed to its original form. This form of compression is often used on images and sound files, for example **jpeg** and **mp3** formats. For image files the number of colors might be reduced and for sound files the number of samples. This type of compression can result in loss of quality.\n",
    "\n",
    "An example of image lossy compression is **chroma subsampling**, color information is reduced by averaging sets of adjacent color values and setting them all to the same value.\n",
    "\n",
    "Audio compression algorithms remove sounds that are outside of human hearing capacity, this technique is called **perceptual audio coding**.\n",
    "\n",
    "#### Lossless Compression\n",
    "With **lossless compression** methods, the original file can be reconstructed from the compressed file. In a nutshell, these methods recognize repeated patterns, replace them with a smaller placeholder and then keep the actual pattern in one place to be later substituted when the file is uncompressed.\n",
    "\n",
    "### Compression tools\n",
    "#### The `gzip` compression tool\n",
    "The name **gzip** refers to both the tool that does the compression and the file format of the resulting file. The compression algorithm used by gzip is known **DEFLATE**. The two main advantages of DEFLATE are speed, and memory efficiency, during the compression process. It only compresses a single file and the resulting file is given a **.gz** file extension.\n",
    "\n",
    "#### The `bzip2` compression tool\n",
    "The **bzip2** compression tool uses an algorithm called **Burrows–Wheeler transform** (**BWT**), another name for it is **block-sorting compression**. Compared to the gzip compression utility, bzip2 ouputs a smaller compressed file but due to the complexity of the algorithm, it takes much longer for the process to complete. It also requires more memory during the compression process than gzip. It only compresses a single file and the resulting file is given a **.bz2** file extension.\n",
    "\n",
    "#### The `xz` compression tool\n",
    "The **xz** compression tool is the newest of the compression tools and uses the **LZMA2** algorithm. It has a greater compression ratio than the **gzip** and **bzip2**. Like bzip2, the greater compression ability comes at the expense of the speed of the process (in some cases it can take 4-5 times longer than bzip2). It only compresses a single file and the resulting file is given a **.xz** file extension.\n",
    "\n",
    "### Archiving\n",
    "**Archiving** is the process of combining multiple files or directories into one single file. Archiving is useful for when we are backing data up or sharing it. For example, if we have multiple files we want to send to someone it is more efficient to turn them into a single archive, send them one file and they can extract the files when they receive it.\n",
    "\n",
    "The word `tar` is short for **Tape Archive** and the command refers back to the early days when backups were made to magnetic tape drives.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Filesystems and Journaling"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A **file system** controls how data is written to, and read from, the location where it is stored. **Journaling** refers to a log that is kept during disk writing that can be used to reconstruct data that might get corrupted due to a power outage or system crash.\n",
    "\n",
    "There are many popular Linux file systems, the user experience on each of them is similar but their underlying features may differ. These are:\n",
    "\n",
    "* **Ext4** is the most popular Linux file system. It is stable and backward compatible with previous versions. It is a journaled system.\n",
    "\n",
    "\n",
    "* The **XFS** file system is known for performance and its ability to handle large files. XFS uses metadata journaling as opposed to journaling both the metadata and data as Ext4 does. This leads to faster performance but can result in loss of data in cases of power loss.\n",
    "\n",
    "\n",
    "* **Btrfs**, short for **B-tree file system**, has built in file system level compression support. The Btrfs file system keeps a checksum of the data stored. It can detect data corruption and recover the corrupted file. Instead of journaling, the Btrfs system employs **Copy-on-Write** (**CoW**). **COW** is an efficiency - in that a copy of data is only made when a change is made, otherwise the copy is a reference to the original data. The Btrfs file system allows us to snapshot the system and save it as a read only file. Btrfs has file system level deduplication, it will remove duplicate copies of data from the system thereby saving space."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Readings"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "* https://levelup.gitconnected.com/how-to-design-a-system-to-scale-to-your-first-100-million-users-4450a2f9703d\n",
    "* https://github.com/madd86/awesome-system-design"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "# Design Facebook News Feed"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Questions to Ask"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "#### Question 1\n",
    "Q: Facebook News Feed consists of multiple major features, like loading a user's news feed, interacting with it (i.e., posting status updates, liking posts, etc.), and updating it in real time (i.e., adding new status updates that are being posted to the top of the feed, in real time). What part of Facebook News Feed are we designing exactly?\n",
    "\n",
    "A: We're designing the core functionality of the feed itself, which we'll define as follows: loading a user's news feed and updating it in real time, as well as posting status updates. But for posting status updates, we don't need to worry about the actual API or the type of information that a user can post; we just want to design what happens once an API call to post a status update has been made. Ultimately, we primarily want to design the feed generation/refreshing piece of the data pipeline (i.e, how/when does it get constructed, and how/when does it get updated with new posts).\n",
    "\n",
    "#### Question 2\n",
    "Q: To clarify, posts on Facebook can be pretty complicated, with pictures, videos, special types of status updates, etc.. Are you saying that we're not concerned with this aspect of the system? For example, should we not focus on how we'll be storing this type of information?\n",
    "\n",
    "A: That's correct. For the purpose of this question, we can treat posts as opaque entities that we'll certainly want to store, but without worrying about the details of the storage, the ramifications of storing and serving large files like videos, etc..\n",
    "\n",
    "#### Question 3\n",
    "Q: Are we designing the relevant-post curation system (i.e., the system that decides what posts will show up on a user's news feed)?\n",
    "\n",
    "A: No. We're not designing this system or any ranking algorithms; you can assume that you have access to a ranking algorithm that you can simply feed a list of relevant posts to in order to generate an actual news feed to display.\n",
    "\n",
    "#### Question 4\n",
    "Q: Are we concerned with showing ads in a user's news feed at all? Ads seem like they would behave a little bit differently than posts, since they probably rely on a different ranking algorithm.\n",
    "\n",
    "A: You can treat ads as a bonus part of the design; if you find a way to incorporate them in, great (and yes, you'd have some other ads-serving algorithm to determine what ads need to be shown to a user at any point in time). But don't focus on ads to start.\n",
    "\n",
    "#### Question 5\n",
    "Q: Are we serving a global audience, and how big is our audience?\n",
    "\n",
    "A: Yes -- we're serving a global audience, and let's say that the news feed will be loaded in the order of 100 million times a day, by 100 million different users, with 1 million new status updates posted every day.\n",
    "\n",
    "#### Question 6\n",
    "Q: How many friends does a user have on average? This is important to know, since a user's status updates could theoretically have to show up on all of the user's friends' news feeds at once.\n",
    "\n",
    "A: You can expect each user to have, on average, 500 friends on the social network. You can treat the number of friends per user as a bell-shaped distribution, with some users who have very few friends, and some users who have a lot more than 500 friends.\n",
    "\n",
    "#### Question 7\n",
    "Q: How quickly does a status update have to appear on a news feed once it's posted, and is it okay if this varies depending on user locations with respect to the location of the user submitting a post?\n",
    "\n",
    "A: When a user posts something, you probably want it to show up on other news feeds fairly quickly. This speed can indeed vary depending on user locations. For instance, we'd probably want a local friend within the same region to see the new post within a few seconds, but we'd likely be okay with a user on the other side of the world seeing the same post within a minute.\n",
    "\n",
    "#### Question 8\n",
    "Q: What kind of availability are we aiming for?\n",
    "\n",
    "A: Your design shouldn't be completely unavailable from a single machine failure, but this isn't a high availability requirement. However, posts shouldn't ever just disappear. Once the user’s client gets confirmation that the post was created, you cannot lose it."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Gathering System Requirements"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "We're designing the core user flow of the Facebook News Feed. This consists of loading a user's news feed, scrolling through the list of posts that are relevant to them, posting status updates, and having their friends' news feeds get updated in real time. We're Specifically designing the pipeline that generates and serves news feeds and the system that handles what happens when a user posts and news feeds have to be updated.\n",
    "\n",
    "We're dealing with about 1 billion users, each with 500 friends on average.\n",
    "\n",
    "Getting a news feed should feel fairly instant, and creating a post should update all of a user's friends' news feeds within a minute. We can have some variance with regards to feed updates depending on user locations.\n",
    "\n",
    "Additionally, we can't be satisfied with a single cluster serving everyone on earth because of large latencies that would occur between that cluster and the user in some parts of the world, so we need a mechanism to make sure the feed gets updated within a minute in the regions other than the one the post was created in.\n",
    "\n",
    "We can assume that the ranking algorithms used to generate news feeds with the most relevant posts is taken care of for us by some other system that we have access to.\n",
    "\n",
    "We'll start with the extremities of our system and work inward, first talking about the two API calls, CreatePost and GetNewsFeed, then, getting into the feed creation and storage strategy, our cross-region design, and finally tying everything together in a fast and scalable way."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### CreatePost API"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "For the purpose of this design, the CreatePost API call will be very simple and look something like this:\n",
    "```\n",
    "    CreatePost(\n",
    "        user_id: string,\n",
    "        post: data\n",
    "    )\n",
    "```\n",
    "When a user creates a post, the API call goes through some load balancing before landing on one of many API servers (which are stateless). Those API servers then create a message on a Pub/Sub topic, notifying its subscribers of the new post that was just created. Those subscribers will do a few things, so let's call them S1 for future reference. Each of the subscribers S1 reads from the topic and is responsible for creating the facebook post inside a relational database.\n",
    "\n",
    "We can have one main relational database to store most of our system's data, including posts and users. This database will have very large tables."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### GetNewsFeed API"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "The GetNewsFeed API call will most likely look like this:\n",
    "```\n",
    "    GetNewsFeed(\n",
    "        user_id: string,\n",
    "        pageSize: integer,\n",
    "        nextPageToken: integer,\n",
    "    ) => (\n",
    "        posts: []{\n",
    "            user_id: string,\n",
    "            post_id: string,\n",
    "            post: data,\n",
    "        },\n",
    "        nextPageToken: string,\n",
    "    )\n",
    "```\n",
    "The pageSize and nextPageToken fields are used to paginate the newsfeed; pagination is necessary when dealing with large amounts of listed data, and since we'll likely want each news feed to have up to 1000 posts, pagination is very appropriate here."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Feed Creation And Storage"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "Since our databases tables are going to be so large, with billions of millions of users and tens of millions of posts every week, fetching news feeds from our main database every time a GetNewsFeed call is made isn't going to be ideal. We can't expect low latencies when building news feeds from scratch because querying our huge tables takes time, and sharding the main database holding the posts wouldn't be particularly helpful since news feeds would likely need to aggregate posts across shards, which would require us to perform cross-shard joins when generating news feeds; we want to avoid this.\n",
    "\n",
    "Instead, we can store news feeds separately from our main database across an array of shards. We can have a separate cluster of machines that can act as a proxy to the relational database and be in charge of aggregating posts, ranking them via the ranking algorithm that we're given, generating news feeds, and sending them to our shards every so often (every 5, 10, 60 minutes, depending on how often we want news feeds to be updated).\n",
    "\n",
    "If we average each post at 10kB, and a newsfeed comprises of the top 1000 posts that are relevant to a user, that's 10MB per user, or 10 000TB of data total. We assume that it's loaded 10 times per day per user, which averages at 10k QPS for the newsfeed fetching.\n",
    "\n",
    "Assuming 1 billion news feeds (for 1 billion users) containing 1000 posts of up to 10 KB each, we can estimate that we'll need 10 PB (petabytes) of storage to store all of our users' news feeds. We can use 1000 machines of 10 TB each as our news-feed shards.\n",
    "```\n",
    "  ~10 KB per post\n",
    "  ~1000 posts per news feed\n",
    "  ~1 billion news feeds\n",
    "  ~10 KB * 1000 * 1000^3 = 10 PB = 1000 * 10 TB\n",
    "```\n",
    "\n",
    "To distribute the newsfeeds roughly evenly, we can shard based on the user id.\n",
    "\n",
    "When a GetNewsFeed request comes in, it gets load balanced to the right news feed machine, which returns it by reading on local disk. If the newsfeed doesn't exist locally, we then go to the source of truth (the main database, but going through the proxy ranking service) to gather the relevant posts. This will lead to increased latency but shouldn't happen frequently."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Wiring Updates Into Feed Creation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "We now need to have a notification mechanism that lets the feed shards know that a new relevant post was just created and that they should incorporate it into the feeds of impacted users.\n",
    "\n",
    "We can once again use a Pub/Sub service for this. Each one of the shards will subscribe to its own topic--we'll call these topics the Feed Notification Topics (FNT)--and the original subscribers S1 will be the publishers for the FNT. When S1 gets a new message about a post creation, it searches the main database for all of the users for whom this post is relevant (i.e., it searches for all of the friends of the user who created the post), it filters out users from other regions who will be taken care of asynchronously, and it maps the remaining users to the FNT using the same hashing function that our GetNewsFeed load balancers rely on.\n",
    "\n",
    "For posts that impact too many people, we can cap the number of FNT topics that get messaged to reduce the amount of internal traffic that gets generated from a single post. For those big users we can rely on the asynchronous feed creation to eventually kick in and let the post appear in feeds of users whom we've skipped when the feeds get refreshed manually."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Cross-Region Strategy"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "When CreatePost gets called and reaches our Pub/Sub subscribers, they'll send a message to another Pub/Sub topic that some forwarder service in between regions will subscribe to. The forwarder's job will be, as its name implies, to forward messages to other regions so as to replicate all of the CreatePost logic in other regions. Once the forwarder receives the message, it'll essentially mimic what would happen if that same CreatePost were called in another region, which will start the entire feed-update logic in those other regions. We can have some additional logic passed to the forwarder to prevent other regions being replicated to from notifying other regions about the CreatePost call in question, which would lead to an infinite chain of replications; in other words, we can make it such that only the region where the post originated from is in charge of notifying other regions.\n",
    "\n",
    "Several open-source technologies from big companies like Uber and Confluent are designed in part for this kind of operation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### System Diagram"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "<img src='imgs/facebook_news_feed.png' alt='facebook_news_feed' width=1000 height=700>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "# Design Instagram News Feed"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "**Similar Services: Twitter Newsfeed, Quora Newsfeed, Facebook Newsfeed**\n",
    "\n",
    "A Newsfeed is the constantly updating list of stories in the middle of Instagram’s homepage. It includes status updates, photos, videos, links, app activity, and ‘likes’ from people, pages, and groups that a user follows on Instagram. In other words, it is a compilation of a complete scrollable version of our friends’ and our life story from photos, videos, locations, status updates, and other activities.\n",
    "\n",
    "For any social media site we design - Twitter, Instagram, or Facebook - we will need some newsfeed system to display updates from friends and followers."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Requirements and Goals of the System"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "#### Functional requirements:\n",
    "* Newsfeed will be generated based on the posts from the people, pages, and groups that a user follows.\n",
    "* A user may have many friends and follow a large number of pages/groups.\n",
    "* Feeds may contain images, videos, or just text.\n",
    "* Our service should support appending new posts as they arrive to the newsfeed for all active users.\n",
    "\n",
    "#### Non-functional requirements:\n",
    "* Our system should be able to generate any user’s newsfeed in real-time - maximum latency seen by the end user would be 2s.\n",
    "* A post shouldn’t take more than 5s to make it to a user’s feed assuming a new newsfeed request comes in."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Capacity Estimation and Constraints"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "Let’s assume on average a user has 300 friends and follows 200 pages.\n",
    "\n",
    "Traffic estimates: Let’s assume 300M daily active users with each user fetching their timeline an average of five times a day. This will result in 1.5B newsfeed requests per day or approximately 17,500 requests per second.\n",
    "\n",
    "Storage estimates: On average, let’s assume we need to have around 500 posts in every user’s feed that we want to keep in memory for a quick fetch. Let’s also assume that on average each post would be 1KB in size. This would mean that we need to store roughly 500KB of data per user. To store all this data for all the active users we would need 150TB of memory. If a server can hold 100GB we would need around 1500 machines to keep the top 500 posts in memory for all active users."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### System APIs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "Once we have finalized the requirements, it’s always a good idea to define the system APIs. This should explicitly state what is expected from the system.\n",
    "\n",
    "We can have SOAP or REST APIs to expose the functionality of our service. The following could be the definition of the API for getting the newsfeed:\n",
    "```\n",
    "getUserFeed(api_dev_key, user_id, since_id, count, max_id, exclude_replies)\n",
    "```\n",
    "Parameters:\n",
    "* **api_dev_key** (string): The API developer key of a registered can be used to, among other things, throttle users based on their allocated quota.\n",
    "* **user_id** (number): The ID of the user for whom the system will generate the newsfeed.\n",
    "* **since_id** (number): Optional; returns results with an ID higher than (that is, more recent than) the specified ID.\n",
    "* **count** (number): Optional; specifies the number of feed items to try and retrieve up to a maximum of 200 per distinct request.\n",
    "* **max_id** (number): Optional; returns results with an ID less than (that is, older than) or equal to the specified ID.\n",
    "* **exclude_replies**(boolean): Optional; this parameter will prevent replies from appearing in the returned timeline.\n",
    "\n",
    "Returns: (JSON) Returns a JSON object containing a list of feed items."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Database Design"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "There are three primary objects: User, Entity (e.g. page, group, etc.), and FeedItem (or Post). Here are some observations about the relationships between these entities:\n",
    "\n",
    "* A User can follow other entities and can become friends with other users.\n",
    "* Both users and entities can post FeedItems which can contain text, images, or videos.\n",
    "* Each FeedItem will have a UserID which will point to the User who created it. For simplicity, let’s assume that only users can create feed items, although, on Instagram Pages can post feed item too.\n",
    "* Each FeedItem can optionally have an EntityID pointing to the page or the group where that post was created.\n",
    "\n",
    "If we are using a relational database, we would need to model two relations: User-Entity relation and FeedItem-Media relation. Since each user can be friends with many people and follow a lot of entities, we can store this relation in a separate table. The “Type” column in “UserFollow” identifies if the entity being followed is a User or Entity. Similarly, we can have a table for FeedMedia relation.\n",
    "\n",
    "<img src='imgs/instagram_database.png' alt='instagram_database' width=600 height=700>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### High Level System Design"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "At a high level this problem can be divided into two parts:\n",
    "\n",
    "**Feed generation**: Newsfeed is generated from the posts (or feed items) of users and entities (pages and groups) that a user follows. So, whenever our system receives a request to generate the feed for a user (say Jane), we will perform the following steps:\n",
    "\n",
    "* Retrieve IDs of all users and entities that Jane follows.\n",
    "* Retrieve latest, most popular and relevant posts for those IDs. These are the potential posts that we can show in Jane’s newsfeed.\n",
    "* Rank these posts based on the relevance to Jane. This represents Jane’s current feed.\n",
    "* Store this feed in the cache and return top posts (say 20) to be rendered on Jane’s feed.\n",
    "* On the front-end, when Jane reaches the end of her current feed, she can fetch the next 20 posts from the server and so on.\n",
    "\n",
    "One thing to notice here is that we generated the feed once and stored it in the cache. What about new incoming posts from people that Jane follows? If Jane is online, we should have a mechanism to rank and add those new posts to her feed. We can periodically (say every five minutes) perform the above steps to rank and add the newer posts to her feed. Jane can then be notified that there are newer items in her feed that she can fetch.\n",
    "\n",
    "**Feed publishing**: Whenever Jane loads her newsfeed page, she has to request and pull feed items from the server. When she reaches the end of her current feed, she can pull more data from the server. For newer items either the server can notify Jane and then she can pull, or the server can push, these new posts.\n",
    "\n",
    "At a high level, we will need following components in our Newsfeed service:\n",
    "* **Web servers**: To maintain a connection with the user. This connection will be used to transfer data between the user and the server.\n",
    "* **Application server**: To execute the workflows of storing new posts in the database servers. We will also need some application servers to retrieve and to push the newsfeed to the end user.\n",
    "* **Metadata database and cache**: To store the metadata about Users, Pages, and Groups.\n",
    "* **Posts database and cach**e: To store metadata about posts and their contents.\n",
    "* **Video and photo storage, and cache**: Blob storage, to store all the media included in the posts.\n",
    "* **Newsfeed generation service**: To gather and rank all the relevant posts for a user to generate newsfeed and store in the cache. This service will also receive live updates and will add these newer feed items to any user’s timeline.\n",
    "* **Feed notification service**: To notify the user that there are newer items available for their newsfeed.\n",
    "\n",
    "Following is the high-level architecture diagram of our system. User B and C are following User A.\n",
    "\n",
    "<img src='imgs/instagram_design.png' alt='instagram_design' width=900 height=700>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Detailed Component Design"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "#### Feed generation\n",
    "\n",
    "Let’s take the simple case of the newsfeed generation service fetching most recent posts from all the users and entities that Jane follows; the query would look like this:\n",
    "```\n",
    "(SELECT FeedItemID FROM FeedItem WHERE UserID in (\n",
    "    SELECT EntityOrFriendID FROM UserFollow WHERE UserID = <current_user_id> and type = 0(user))\n",
    ")\n",
    "UNION\n",
    "(SELECT FeedItemID FROM FeedItem WHERE EntityID in (\n",
    "    SELECT EntityOrFriendID FROM UserFollow WHERE UserID = <current_user_id> and type = 1(entity))\n",
    ")\n",
    "ORDER BY CreationDate DESC \n",
    "LIMIT 100\n",
    "```\n",
    "\n",
    "Here are issues with this design for the feed generation service:\n",
    "* Crazy slow for users with a lot of friends/follows as we have to perform sorting/merging/ranking of a huge number of posts.\n",
    "* We generate the timeline when a user loads their page. This would be quite slow and have a high latency.\n",
    "* For live updates, each status update will result in feed updates for all followers. This could result in high backlogs in our Newsfeed Generation Service.\n",
    "* For live updates, the server pushing (or notifying about) newer posts to users could lead to very heavy loads, especially for people or pages that have a lot of followers. To improve the efficiency, we can pre-generate the timeline and store it in a memory.\n",
    "\n",
    "#### Offline generation for newsfeed\n",
    "We can have dedicated servers that are continuously generating users’ newsfeed and storing them in memory. So, whenever a user requests for the new posts for their feed, we can simply serve it from the pre-generated, stored location. Using this scheme, user’s newsfeed is not compiled on load, but rather on a regular basis and returned to users whenever they request for it.\n",
    "\n",
    "Whenever these servers need to generate the feed for a user, they will first query to see what was the last time the feed was generated for that user. Then, new feed data would be generated from that time onwards. We can store this data in a hash table where the “key” would be UserID and “value” would be a STRUCT like this:\n",
    "```\n",
    "Struct {\n",
    "    LinkedHashMap<FeedItemID, FeedItem> feedItems;\n",
    "    DateTime lastGenerated;\n",
    "}\n",
    "```\n",
    "We can store FeedItemIDs in a data structure similar to Linked HashMap or TreeMap, which can allow us to not only jump to any feed item but also iterate through the map easily. Whenever users want to fetch more feed items, they can send the last FeedItemID they currently see in their newsfeed, we can then jump to that FeedItemID in our hash-map and return next batch/page of feed items from there.\n",
    "\n",
    "#### How many feed items should we store in memory for a user’s feed? \n",
    "Initially, we can decide to store 500 feed items per user, but this number can be adjusted later based on the usage pattern. For example, if we assume that one page of a user’s feed has 20 posts and most of the users never browse more than ten pages of their feed, we can decide to store only 200 posts per user. For any user who wants to see more posts (more than what is stored in memory), we can always query backend servers.\n",
    "\n",
    "#### Should we generate (and keep in memory) newsfeeds for all users? \n",
    "There will be a lot of users that don’t log-in frequently. Here are a few things we can do to handle this:\n",
    "* a more straightforward approach could be, to use an LRU based cache that can remove users from memory that haven’t accessed their newsfeed for a long time \n",
    "* a smarter solution can figure out the login pattern of users to pre-generate their newsfeed, e.g., at what time of the day a user is active and which days of the week does a user access their newsfeed? etc.\n",
    "\n",
    "#### Live updates/ Feed publishing\n",
    "The process of pushing a post to all the followers is called **fanout**. By analogy, the push approach is called fanout-on-write, while the pull approach is called fanout-on-load. Let’s discuss different options for publishing feed data to users.\n",
    "\n",
    "* **\"Pull\" model or Fan-out-on-load**: This method involves keeping all the recent feed data in memory so that users can pull it from the server whenever they need it. Clients can pull the feed data on a regular basis or manually whenever they need it. Possible problems with this approach are a) New data might not be shown to the users until they issue a pull request, b) It’s hard to find the right pull cadence, as most of the time pull requests will result in an empty response if there is no new data, causing waste of resources.\n",
    "\n",
    "* **\"Push\" model or Fan-out-on-write**: For a push system, once a user has published a post, we can immediately push this post to all the followers. The advantage is that when fetching feed you don’t need to go through your friend’s list and get feeds for each of them. It significantly reduces read operations. To efficiently handle this, users have to maintain a Long Poll request with the server for receiving the updates. A possible problem with this approach is that when a user has millions of followers (a celebrity-user) the server has to push updates to a lot of people.\n",
    "\n",
    "* **Hybrid**: An alternate method to handle feed data could be to use a hybrid approach, i.e., to do a combination of fan-out-on-write and fan-out-on-load. Specifically, we can stop pushing posts from users with a high number of followers (a celebrity user) and only push data for those users who have a few hundred (or thousand) followers. For celebrity users, we can let the followers pull the updates. Since the push operation can be extremely costly for users who have a lot of friends or followers, by disabling fanout for them, we can save a huge number of resources. Another alternate approach could be that, once a user publishes a post, we can limit the fanout to only her online friends. Also, to get benefits from both the approaches, a combination of ‘push to notify’ and ‘pull for serving’ end-users is a great way to go. Purely a push or pull model is less versatile.\n",
    "\n",
    "#### How many feed items can we return to the client in each request?\n",
    "We should have a maximum limit for the number of items a user can fetch in one request (say 20). But, we should let the client specify how many feed items they want with each request as the user may like to fetch a different number of posts depending on the device (mobile vs. desktop).\n",
    "\n",
    "#### Should we always notify users if there are new posts available for their newsfeed?\n",
    "It could be useful for users to get notified whenever new data is available. However, on mobile devices, where data usage is relatively expensive, it can consume unnecessary bandwidth. Hence, at least for mobile devices, we can choose not to push data, instead, let users “Pull to Refresh” to get new posts."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Feed Ranking"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "The most straightforward way to rank posts in a newsfeed is by the creation time of the posts, but today’s ranking algorithms are doing a lot more than that to ensure “important” posts are ranked higher. The high-level idea of ranking is first to select key “signals” that make a post important and then to find out how to combine them to calculate a final ranking score.\n",
    "\n",
    "More specifically, we can select features that are relevant to the importance of any feed item, e.g., number of likes, comments, shares, time of the update, whether the post has images/videos, etc., and then, a score can be calculated using these features. This is generally enough for a simple ranking system. A better ranking system can significantly improve itself by constantly evaluating if we are making progress in user stickiness, retention, ads revenue, etc."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true,
    "hidden": true
   },
   "source": [
    "### Data Partitioning"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "hidden": true
   },
   "source": [
    "#### Sharding posts and metadata\n",
    "Since we have a huge number of new posts every day and our read load is extremely high too, we need to distribute our data onto multiple machines such that we can read/write it efficiently. For sharding our databases that are storing posts and their metadata, we can have a similar design as discussed under Designing Twitter.\n",
    "\n",
    "#### Sharding feed data\n",
    "For feed data, which is being stored in memory, we can partition it based on UserID. We can try storing all the data of a user on one server. When storing, we can pass the UserID to our hash function that will map the user to a cache server where we will store the user’s feed objects. Also, for any given user, since we don’t expect to store more than 500 FeedItemIDs, we will not run into a scenario where feed data for a user doesn’t fit on a single server. To get the feed of a user, we would always have to query only one server. For future growth and replication, we must use Consistent Hashing."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "hidden": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "hidden": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "hidden": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.7"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}