System design notes

This is a detailed walkthrough of the system design course at InterviewReady. If you are looking to prepare for interviews, this walkthrough will help. The walkthrough describes how high-level, low level and fundamental concepts of design are separated into sections inside the course. The platform includes note-taking, quizzes, architecture diagrams, and discussion forums.

Back-of-the-envelope calculations are often expected in system design questions. They help logically state the parameters influencing a result, and estimating the capacity requires multiple estimations on the way. Also lets us individually state our assumptions. Eg: Estimate the hardware requirements to set up a system like YouTube. Eg: Estimate the number of petrol pumps in the city of Mumbai. Chapters 00:06 Storage Requirements 01:20 Supplementary storage requirements 03:54 Back of Envelope calculations 05:38 Youtube caching estimation 08:58 Youtube video processing estimation 12:14 Conclusion ------STORAGE Let's start with storage requirements: About 1 billion active users. I assume 1/1000 produces a video a day. Which means 1 million new videos a day. What's the size of each video? Assume the average length of a video to be 10 minutes. Assume a 10 minute video to be of size 1 GB. Or... A video is a bunch of images. 10 minutes is 600 seconds. Each second has 24 frames. So a video has 25*600 = 150,000 frames. Each frame is of size 1 MB. Which means (1.5 10^5) (10^6) bytes = 150 GB. This estimate is very inaccurate, and hence we must either revise our estimate or hope the interviewer corrects us. Normal video of 10 minutes is about 700 MB. As each video is of about 1GB, we assume the storage requirement per day is 1GB * 1 million = 1 PB. This is the bare minimum storage requirement to store the original videos. If we want to have redundancy for fault tolerance and performance, we have to store copies. I'll choose 3 copies. That's 3 petabytes of raw data storage. What about video formats and encoding? Let's assume a single type of encoding, mp4, and the formats will take a 720p video and store it in 480, 360, 240 and 144p respectively. That means approximately half the video size per codec. If X is the original storage requirement = 1 PB, We have X + X/2 + X/4 + X/8 == 2*X. With redundancy, that's 2X 3 = 6 X. That's 6 PB(processed) + 3PB (raw) == 10 PB of data. About 100 hard drives. The cost of this system is about 1 million per day. For a 3 year plan, we can expect a 1 billion dollar storage price. Now let's look at the real numbers: Video upload speed = 3 * 10^4 minutes per minute. That's 3 * 10^4 * 1440 video footage per day = 4.5 * 10^7 minutes. Video encoding can reduce a 1-hour film to 1 GB. So 1 million GB is the requirement. That's 1 PB. So the original cost is similar to what the real numbers say. If we are off by order of magnitude, it's good. However, being off by 3 or more orders of magnitude is too much. We can then highlight the following: Where our assumption was wrong, or Which factor we didn't take into account.

Here are 5 Tips for System Design interviews. They are helpful when preparing for a System Design interview. 1. Don't get into details prematurely 2. Avoid fitting requirements to a set architecture in mind 3. Keep it simple, stupid! Remember to look at the big picture and avoid too many hacks when solving. 4. Have justifications for the points you make. Don't use buzz words or half hearted thoughts in your design. 5. Be aware of the current solutions and tech practices. A lot of solutions can be purchased off the shelf which simplify implementation. You should be able to argue for a custom implementation with it's pros and cons. Have these on the back of your mind during you interview, and all the best! Here are three major points evaluated during the interview: 1. Clarity of Thought a. Express your thoughts in a clear manner. b. Justify your decisions. Critical reasoning and argument are key to a successful software design. c. When faced with a problem, use standard approaches to mitigate it. For example, say you are faced with an availability problem. State that replication and partitioning help increase availability in general, and move on to offer a solution. d. Don’t make points without thinking them through. Half-hearted attempts at solving problems are frowned upon heavily. 2. Know about existing solutions a. Stay up to date with the current solutions in the market. This includes products and design practices. If NoSQL is being adopted left right and center, you need to be aware of it. b. Know when to pick a solution vs. building something custom. If you name a product, you should be (generally) aware of the features it provides. c. Design practices enable you to meet custom requirements. Examples are decoupling systems, load balancing, sticky sessions, etc… 3. Flexibility a. Switch your targets as the requirements shift. If the interviewer wants to know about one particular part of the system, do it first. b. Never have a set architecture in mind. We all try to fit requirements to a system, but only after it has been shaped by the initial ones. A rigid attitude creates a brittle architecture. It will break before you do. c. Take a step back at times to make adjustments to the general architecture. Being focused on one part can narrow our vision and bloat those areas. There will be components which can be extracted out and extended to the rest of the system.

HTTP, WebSockets, TCP and UDP This video is everything you need to know about Hypertext Transfer Protocols. Let me ask you something. Can a server send a client request to the client by itself? Do you know about the web sockets? Wanna know about Transmission Control Protocol(TCP)? Get diving straight into the video folks.

What was cooking up interesting in the zoom class?? What's life without exploring? In this zoom class let us explore Kafka! How many of us know about the Kafka internals? Kafka is ubiquitous. The questions that come up to our mind when we want to dig deep into Kafka are: How is Kafka Internally built? It's a log. A log is a file where u write one line and when there is a new event you write another line. Every line will have a delimiter. Every line is appended after the delimiter. Why Kafka doesn't work on queues(priority)? Queues are implemented using linked lists or an array. The mechanism of Kafka doesn't give any priority. Kafka is fast, scalable, and distributed in nature by its design, partitioned and replicated commit log service. So there is no priority on topic or message. All events belonging to a topic need to be ordered. For example, Let's say we have topics A B C and P Q R. For instance, if all this is jumbled like P Q A B C R Rules: P comes before Q And Q comes before R The original order need not be maintained. We need to care about ordering between inside the topics not in between topics. Wherein C will be consumed by the consumer only after B gets consumed. What is Causal ordering? As long as relevant messages are sent to the same partition, message (causal) ordering is still guaranteed. In Kafka, this is achieved using keyed messages. Kafka partitioners would hash keys and ensure a key always goes to the same partition given the number of partitions is not changed. How do u search in a log? A log has a key and value. The log is already sorted by key. How we usually query is based on timestamp or the machine id. Instead of taking any random key, we take the 128 bits key and use 64 bits to store the timestamp. For Example: 18th August_123 Where ‘_’ is used as a delimiter. Elastic search The basic idea is it is a no-sequel database. It has key-value pairs. It is good at text search and aggregations. You can add tags to it. An Elasticsearch index is a collection of documents that are related to each other. Benefits of using Kafka Limited persistence Fast in terms of writing(if u want to read fast use another database). Used for debugging Finally, If a very popular search text exists how do u find the document that it exists in and store it efficiently? I leave it here for you to discover. Let me give out a little secret! This zoom class has the answer to this question!

In this zoom video, get ready to start with interview-ready tips. Let's say there are 5-6 servers and there is a load balancer... How are the requests monitored? If we have authentication in the load balancer it becomes a gateway that checks whether a request is valid or not. If we need extra authentication needed( for example, only admins can send requests to a particular server), the question arises of where we should have the authentication logic. If you are up for scaling then you must be opting Load Balancer Authentication. The drawback of this method is that you will have to deploy a load balancer every time a request is sent. Another drawback is that it is a single point of failure. Who will handle authentication when the Load balancer is gotten rid of? AuthService We have an Auth service which takes in request from load balance them validates and send them back. What if I tell you we have a video to explain this? Go ahead and watch the video on Authentication and Global caching under the High-Level Design. Service Registry If we are going to have a sidecar to take care of invalidation, how do you maintain the states? The consistency becomes a question. The state here is the cache that we use. What is a sticky session? Dive into the video on time stamp 23: 50 to know more. Authentication for multiple systems works in different ways. The architecture-wise where you give the authentication matters in terms of performance. You have a bunch of externally exposed services one of the things you can do is if it is coming from the service which is sending the request can say whether it needs to be authenticated or not. If internal services are not talking to each other and the sidecar feels that the request need not be authenticated, the authentication can be bypassed. Benefits Of Compression: Lower bandwidth usage Faster to send Lesser memory being used Why shouldn't we compress messages? Computational overhead Time-taking latency. To know how to implement a file system in an interview dive into the time stamp- 1:19:00 and keep your knowledge caps open. For this problem, we need to be creating : 1. Create(file_id, content) 2. Read(file_id) 3. Update(file_id, content) 4. Delete(file_id) It's time for discovery. Go ahead, explorer. The zoom class answers this question!

How to count the frequency of words from 1TB of file in 2 GB RAM consistency having only a single machine? There are 2 things that could be done in this case. To split the whole file into contents of 2 GB. The other approach would be to stream the file while tokenizing it and counting the number of words until it reaches the required size. Data Structures that can be used: Hash Maps The entire words of the file will be converted into a key set of 1 Terabyte long if there are distinct words in the file. The edge would be storing 1 next to all the hashed words of the file which will be altered with changing frequencies of the words. We might need concurrent hashmaps to deal with this problem The scatter-and-gather approach can be used to sort out the data. Have a look at the external Merge Sort algorithm (Which can also be used to solve this problem) Splitting the file into pieces is the complexity. Multi-threading can be used to solve this problem. Let's say we have a fileSplitter(n), There are 1,2,3 threads that are being passed. The splitter will split the file into pieces based on the number of words such that the limit memory is reached. We will have to stream through these chunks of threads. We will need to have seek() and parse() functions to go ahead with this method. Finally, we store all of this in hashmaps. Quick Fact!! BookMyShow makes use of Optimistic locks !! There are 2 copies to make sure your booking doesn't clash with already booked seats. While pessimistic locks won't create copies. The timestamps will help you to reach the point in the video where you can see a full video explanation of the same. Why not solve your doubts? When you are connecting to a server you are connecting to a port where multiple people connect. The server can identify who is connected to it. Let's go back to TCP connections, We take a message and route it through multiple routers and send it to a person. It is more of a logical concept rather than a physical one. Let's Move on to a system design problem: There is an e-commerce app which has a lot of price fluctuations for its products. They have a catalogue of schools that is already in the ERP systems. They want to add a new component to the e-commerce platform. Get to know the tips on how to solve this!! Come lets clear a user's doubt on over-utilizations of resources.

This zoom meeting can never get interesting enough!! Get to know the resources on where to apply for SD roles from this zoom meet intro part.. Some websites that were discussed: Instahyre Cutshort Geektrust Angel.co Along with that why not hustle with a few more resolving doubts and general discussions? How do we handle 5000 transactions(writes) per second on a relational database with a max latency of 50 milliseconds? Moving ahead with the learning spree.. The first thing to note here is what type are consistency are looking at and what type of translation isolation level is needed? Choosing the translation isolation level in a wise manner will help increase concurrency. The requirements for the questions are below: The user should get appropriate tasks to work on based on skill language. Example: A user who only knows Hindi should be assigned Hindi tasks where these tasks should be served in order of priority and first come first serve basis. By default, users should get English tasks as well. The system management should also be able to view the life cycle of a job and we expect the people to be working in parallel. The system must be able to respond to the task within 15 milliseconds if the user is waiting for the job or not( If he is free). Let's take a queue and enter the tasks. This means that we have a priority queue which is implemented by a heap-based on which the tasks are served. Let's say we have multiple queues. The problem to address here is the priority and skill level of the worker and the language might be different. We will have a head-of-line blocking problem when we have the tasks blocking the assigning of work to the users based on priority when using only one queue. We can have a list of queues for the languages task and the queue is following FIFO(as we know) if the task on the front doesn't get assigned to any worker we upload that task to a separate log queue. We use the approach of using a sorted tree instead of a priority queue which makes the problem complexity O(log(n)). This is very similar to the Operating System Resources problem. Now, what happens when we use a hash map(brute force algorithm)? Depending on the language we have a worker. Then we see the skill level of the worker. To remove the problem where 2 workers pull the same task we shall assign locks. Acknowledgements for the acceptance or assignments of the tasks need to be mentioned. Some sort of expiry algorithm would do. Question: There are 2 services A and B deployed using EKS. What is the best way to communicate with each service? Also, where do we need to keep the URL? Go ahead and watch the video to know what is Gaurav’s take on this!!

We discuss how the IRCTC algorithm possibly works. It's a complex algorithm, but worth understanding! Irctc Algorithm Train ticketing system How do we efficiently book tickets in the ticket booking system so the seats don't remain vacant? It's a High-level design question. Example: There are 10 stations and some person X books seats from 1 to 5 and some other person wants to book from 6 to 8. How do we handle that? How do we know who booked prior? Here the exciting part is that we not only have to consider the allocations, but also we will have to search for trains. We draw the UML diagram by considering the following: •There is a Client who wants to book a seat through their web browser. •The clients say that they want to book a journey with the specifications • This is the case when the client wants to book a specific train. • The other option is when the client searches for all trains available to reach the particular destination with the number of seats, and start date. • The train should have information about the estimated time of arrival, estimated departure time, source, destinations, category and seats. Multiple perspectives and multiple calculations..let's head to see the exact answer. How do you book the seats? The other interesting fact is that the centre of gravity has to be taken into account or the train will be heavier on one side alone. If a single woman is travelling she needs to be placed with other women travelling on the same train. All these are the considerations that need to be considered. However, the brute force approach would be: Book from A - B ( 2 seats) The algorithm is like a rule engine. It goes through the preferences and chooses the best option for the client. The time the rule engine runs to check all the preferences will become a matter of consideration here. The duplication is good because it will make the search faster. Instead of a beat matrix, we can flatten it by making it an array. So how do we approach this? Let's head to the video!

Google Docs and Chess Design Queries When we use Google docs when we are sharing a document with someone else. Both will be able to edit the content of the docs. This makes it difficult to design the algorithm as we will have to lock based on the ranges. A specific algorithm is used for these kinds of test collaborations. The Operational Transform The algorithm is very complex. There is a video available in the System Design course for your reference. Let's move to the Chess design: The Chess design is going to have a gateway. The gateway speaks to another service that stores session data which does a protocol translation. In a chess design, the app should make regular changes. Different clients speak to the app to hit the services. The problem: If the API has a breaking change then all the clients need to be updated and restarted. When changing a version of the library you need to update and restart the services. The key here is to keep the gateway dumb as possible so that you do need not to restart the gateway. •Minor changes are not relevant • Services are not relevant • Breaking changes are not relevant The service registry needs to be dumb as possible. Apart from a reverse proxy it also acts as a parser. Let's say you are in version 3 of Google Docs and google docs says that a person deleted a character from the doc that will be stored in a new version. What happens is that one keeps drifting away from the lattice point. Eventually, the events are mapped to the current state. The algorithm will tell you how to do that. Each operation is mapped which will end up in the final result. Further, let's see what exciting discussions we have and how to approach them. Head to watch the full video to learn more!

The process of navigating to a website, specifically mentioning the role of DNS servers and Content Delivery Networks (CDNs) in the process. When a user types in a website address, such as amazon.com, the request first goes to a DNS server which then determines which CDN to direct the request. The CDN acts as a first line of defense against Distributed Denial of Service (DDOS) attacks, and then the request goes to the real server (content server), which delivers the content. Protection from malicious attacks can be done at the CDN or Gateway level. He gives the example of a website hosted on S3, with the CDN being backed by S3 and replicating static content all around the world for low latency. The DNS server can determine the closest location for the user to connect and provide the appropriate IP address.

Gaurav discusses using consistent hashing in sharding and how requests are mapped to specific buckets in some distributed systems like Redis. There is a potential for one bucket to become overloaded due to a lack of uniform distribution. What is the potential latency when consumers pull during a load rebalancing event? Graph API vs. Open API. There are various ways to overcome the problem of an increase in logs, including compressing and archiving logs, scaling the storage for logs, and identifying which service is causing the increase in logs. It is impossible to stop an increase in logs as user activity increases, but steps can be taken to manage them more efficiently. Additionally, discussing with teams why they are logging so much is a practical consideration. The use of APIs and understanding how they work are essential. GraphQL may not be as helpful as REST in certain situations and clients must understand the objects beforehand. When there is a large request load on the API gateway, it can be scaled horizontally to handle the load, but that code can also be made more efficient. YouTube handles a similar scenario, but when there is too much load, the cached content can be used. There is a need to change and degrade the service in the API gateway when there is too much load. Gaurav shares his experience preparing for technical interviews, specifically for DSA. It's difficult to find an equilibrium between system design and DSA prep. It's essential to speak with the interviewer about their thought process and approach to a problem. For DSA, execution is very important, and being able to understand and catch hints is also essential. In an interview, it's best to be prepared and not overlook the importance of paying attention to the current needs. One complex DB query as an API versus using three different simple queries as three APIs: depends on the requirement, but usually, three simple queries are better, but sometimes it may be necessary to use one complex query for an atomic operation. We found a solution for handling large requests and responses for an API. The idea of breaking down a large request into smaller fragments and returning responses is applicable here. The input and decision on how to break down the request can depend on the business use case and that there is a separation of ingestion rate and persistence rate. We may also use an asynchronous approach, where data is uploaded and inserted into a database after all files have been processed and cross-checks can be made on the records.

The video by Gaurav Sen covers a range of topics. The discussion starts with Kafka and SCS, which are message queueing systems. Gaurav suggests some resources for learning about Kafka, including Confluent's documentation. The conversation then moves on to managing permissions and naming conventions for meetings. The video ends with a design problem that involves designing an authentication system for microservices with complex rules for user actions.

In the video, Gaurav Sen talks about the importance of consistency in distributed systems. He gives an example of how likes on a post in a social media platform may not be immediately consistent across all nodes in the system, but it is still acceptable because it does not affect the overall experience. He also explains how consistency can be ensured in financial systems through settlements and reconciliations, and how a sorted string table can be used to maintain a record of changes in the system. Sen emphasizes that the level of consistency required depends on the application and the user's needs.

The video by Gaurav Sen shows a session in which he discusses system design, specifically the difference between machine coding and low level design. He also discusses the types of questions asked during SD interviews and gives an example of a recommendation system. In the latter part of the video, he demonstrates how to solve a system design problem using Redis and hash maps to return the most popular blog posts in the last five and 15 minutes.

The video discusses the issue of ordering events in a message queue or event bus with multiple topics, using examples of services posting response times and events like create, update, and delete. Enforcing order by timestamps or using distributed consensus algorithms like Paxos or Raft is mentioned as a solution. The use of logical clocks like hybrid logical clocks or Lamport clocks is also discussed as an alternative to the Spanner implementation, which requires Google-level hardware. We also discuss the importance of enforcing order in the queue to ensure the sequential execution of instructions in a multi-threaded environment. It explains how a thread pool can mimic a queue and how events can be sent to different threads depending on whether they need to be sequentially ordered. The video also addresses questions related to hashing, partitioning, and handling multiple events across different services. Finally, it suggests that ordering across services can be complex and require engineering coordination.

We attempt multiple questions of the system design judge!

We discuss the architecture and tradeoffs that come up when designing a live watch party system.

Connecting to the internet: ISPs, DNS and everything in between At times do you wonder how your computer connects to the internet? Now that you know the basics of networking, have you ever thought about where you get data from a server? What is the role of an IP address? Wanna know about the Internet backbone? Dive into the video for we are about to learn about Domain Name Servers and Internet Service Providers.

Communication Standards: REST, GraphQL and GRPC In this video you will learn about the following: In a microservice architecture, services need to communicate with each other. How do servers communicate internally? How do you communicate between services? Different services are written in different languages. But why do we need one common language to define the objects? How do you query the microservices? REST GraphQL gRPC (google Remote Procedure Call) is generally used by microservices to communicate internally. It is written over HTTP 2.0

The internal architecture of Cassandra helps us understand some design patterns of NoSQL key-value databases. Sharding Quorum Redundancy Client-side routing with keys

Benefits of a cache: Saves network calls Avoids repeated computations Reduces DB load Drawbacks of a cache: Can be expensive Potential thrashing Cache Policies: LRU LFU

Replacement Policies In this video, we learn to determine which key needs to be evicted. This is based on the replacement policy LRU Policy (Least Recently Used) In this policy, we evict the entry that has not been used for the longest. To implement this policy, we need to add another data point i.e., Last Used. To understand this policy, watch the video where you can understand it with examples. Time interval in which Write-Around Policy Replacement Policies LRU Policy (Least Recently Used) an entry was not used = current_timestamp - LastUsed for that entry. LFU Policy (Least Frequently Used) In this policy, we evict the entry that is used least frequently. Don't forget to watch the video for a better explanation using examples!!

System design interviews can be tricky. Here are 5 mistakes that you should avoid during an interview. The mistakes mentioned here are what I have seen and done in my experience as a software engineer and interviewer. Do not have a premeditated introduction Estimate Capacity Requirements only if required Expand on the network protocol choices (if required, like we do with database choices) Know the internal architecture of systems Highlight design tradeoffs

We identify the key features of the system by noting down its: 1) Services 2) APIs 3) Database schemas

Systems design a procedure by which we define the architecture of a system to satisfy given requirements. It is a technique by which the required amounts of scalability, reliability, performance and consistency are satisfied in real world systems. We discuss how to start with system design and why it is required. The first concept in designing a system is scalability. We discuss the two main approaches to solve this problem: Horizontal scaling and vertical scaling. Horizontal scaling is adding more machines to deal with increasing requirements. These machines handle requests in parallel to improve user experience. Vertical scaling is replacing the current machines with more advanced machines to improve throughput and hence response time. The techniques are used in conjunction in real world systems.

Microservices are a hot topic in system design interviews. It is important to know why we use them instead of monolithic systems. The short answer is: Scalability. The detailed one would be: Advantages: 1) The microservice architecture is easier to reason about/design for a complicated system. 2) They allow new members to train for shorter periods and have less context before touching a system. 3) Deployments are fluid and continuous for each service. 4) They allow decoupling service logic on the basis of business responsibility 5) They are more available as a single service having a bug does not bring down the entire system. This is called a single point of failure. 6) Individual services can be written in different languages. 7) The developer teams can talk to each other through API sheets instead of working on the same repository, which requires conflict resolution. 8) New services can be tested easily and individually. The testing structure is close to unit testing compared to a monolith. Microservices are at a disadvantage to Monoliths in some cases. Monoliths are favorable when: 1) The technical/developer team is very small 2) The service is simple to think of as a whole. 3) The service requires very high efficiency, where network calls are avoided as much as possible. 4) All developers must have context of all services.

Load balancing is distributing work evenly onto a set of resources. In this video, we explain the main reason for load balancing, along with some common algorithms: Round robin Geo-distributed Least Connection

A single point of failure(SPOF) in computing is a critical point in the system whose failure can take down the entire system. A lot of resources and time is spent on removing single points of failure in an architecture/design. Single points of failure often pop up when setting up coordinators and proxies. These services help distribute load and discover services as they come and leave the system. Because of the critical centralized tasks of these services, they are more prone to being SPOFs. One way to mitigate the problem is to use multiple instances of every component in the service. The graph of dependencies then becomes more flexible, allowing the system to resiliently switch to another service instead of failing requests. Another approach is to have backups that allow a quick switch over on failure. The backups are useful in components dealing with data, like databases. Allocating more resources, distributing the system and replication are some ways of mitigating the problem of SPOF. Hence designs include horizontal scaling capabilities and partitioning. It is important to note that the CAP theorem does not allow removing SPOFs if perfect consistency is required.

Content Delivery Networks are a bunch of servers spread across the globe to serve information. These networks are available on rent to deliver static content quickly to nearby users. Some examples of CDNs are Amazon CloudFront and the Akamai CDN. They are (relatively) cheap to rent and have high availability. They also provide pluggable algorithms to invalidate and fetch data. We discuss why content delivery networks are useful, and how they serve data with examples.

Internal routing: MAC addresses and NAT How does an internal router know where to send an external response to? Do you really wanna find out? Dig into the video right now as in this video we are going to learn: Content Delivery Network(CDN) Problems we face if we do not use CDN Decoding the responses from CDNs

Head-of-line blocking in HTTP Head-Of-Line blocking occurs when the message/data packet at the head of the queue cannot move forward due to congestion even if other messages/packets behind this could. HTTP 2.0 solves this problem using Multiplexing. So what is multiplexing? Why HTTP 3.0 uses UDP instead of TCP? The video above answers all your questions!!

Video transmission: WebRTC and HTTP-DASH What do servers do while handling video data? Video files are large and cannot be sent in a single response. They have to be broken into packets. Why HTTP is not good for video transmission? HTTP is stateless. The client has to specify which chunk it wants because the server does not know about the previous requests. Do you want to learn more that includes the working of Conference Video Protocol? Dig into the video right now!

A white paper shows the implementation details and the practical aspects of building a system. Most of the trade-offs in it come from the product requirements. Memcached at Facebook is the example. A request for a profile is an aggregate query that hits multiple shards, so splitting the key space makes things worse. Facebook chose replication. The papers run from number 20 to number one: Monolith, FlexiRaft, Spanner and Minesweeper: real-time recommendations at TikTok, quorums at Meta, a strongly consistent database at Google, and automated root cause analysis. Cassandra, FoundationDB from Apple and Amazon Aurora: three databases. Pregel, Dapper, Chubby, Megastore, Bigtable, MapReduce and the Google File System: graph processing, tracing, distributed locks, storage and data processing at Google. TAO, the graph store from Meta. Memcached, in the top five. Monarch and Gorilla, the time series databases of Google and Facebook. Gorilla is in the top three. Amazon DynamoDB, at number two. Zanzibar, a system by Google, is number one, for its practical optimizations around rate limiting and fault tolerance.

Memcached at Facebook is a common cache that answers the hundreds of key value reads behind a single page. A home page is a tree of posts, comments and replies. A cache used by all engineers needs high adoption, so engineers import a client library, which compresses values and batches requests. It is a look-aside cache. A get that misses goes to MySQL and populates the cache. A set updates the database and deletes the key, because a delete is idempotent. The cache is scaled according to the access pattern: Protocols. Set requests use TCP, so they go in order. Get requests use UDP, which saves 20% in request latency. Replication. Sharding spreads the keys of a page over all nodes. With replication, one node answers the whole page. Rate limiting. Each client keeps a sliding window that is neither too large nor too small. The gutter. Spare instances, around 1% of all nodes, reduce client failures by 99%. Mcsqueal. The keys to delete are added to the MySQL binary log, which is propagated.

Facebook keeps its social graph in memory, as an adjacency list, so that lookups are fast. Memcached is a mature piece of technology at Facebook. Facebook was one of the first companies with a massive graph network, and graph databases are not common. A graph is represented as an adjacency matrix or an adjacency list. The number of users is very large and the matrix is very sparse, so the graph is kept as an adjacency list: A map. The key is a person and the value is the list of that person's friends. Get key Gaurav returns his friends. A map inside it. Checking whether Harsha and Gaurav are friends goes through the list. Stored as a map, the check is order one. A request for a news feed, or for friends having birthdays, goes through the friend list. These queries are made in memory. The objectives are low latency, high availability and ease of use. Very high consistency is not needed. A cache used by one or two teams limits its optimizations to one team.

One homepage load fetches hundreds of key value pairs at once. A homepage shows posts. The posts have comments and likes, the comments have replies, and people click to see who has liked a post. A popular video is cached with its top comments and their top replies. There is no clear structure, because nobody can guess what users will find popular. So the page is not cached as one piece. It is broken into levels: Level one. Notifications and popular posts, which everyone gets to see. Level two. The top 10 comments of a popular post, or the profile and mutual friends of a person to follow. It is cached too, and fetched only when the person clicks. One homepage request fans out into nested calls: 95 keys for most requests and 1,700 for some. Facebook constructs a directed graph of the data. Some fetches run in parallel and some depend on others. Updates follow read after write consistency. A like stays hit after a refresh, and an added comment shows on the post.

A look-aside cache is a cache deployed apart from the database and outside the web server, which calls it. Memcached is an in-memory distributed hash table, and a write to it is a deletion: A get for key 1 returns 10 from the cache. A set of key 1 to 30 removes the entry from the cache and writes 30 to the database. The next get for key 1 is a cache miss. The value 30 is fetched from the database and populated in the cache. So the cache takes two operations, an add and a delete. An engineer does not manage any of this. The code imports a client library, called mcrouter, and calls get, set and delete. The client does serialization, compression, request routing, error handling and request batching. It is deployed along with the code, as a Mailchimp or an AWS client is, and adoption across the organization is very high. Fetching 1,700 items is likely to hit a failure and a retry, so tail latencies are high.

Memcached at Facebook sends a get request over UDP and a set or delete request over TCP. TCP is acknowledgement based, so all writes are ordered. UDP is best effort and connectionless: no connection is created, and TCP connections are limited for a box and cost something. Facebook noticed a 20% reduction in request latency for get requests. A write can then be processed after the read that follows it: Three requests are sent: get key 1, set the key to 30, get key 1. Handled in parallel, the two gets are processed before the set. Read after write consistency breaks. A user likes a post, refreshes the page, and the like is not there. A smart client avoids this with a version number. The home page is version 29 and the like is version 30. A reply from the cache at version 27 is treated as an error on the client, which does not try to recover from it. The client here is the one hosted on the web server, not the UI client.

A sliding window is a rate limit kept on the client: it moves forward as responses come back, which limits the number of requests in flight. The size of the window is a trade-off. A larger window causes more congestion, so the last request waits a long time. A smaller window causes more round trips, because the requests beyond it are rejected and tried again later. The size is therefore not fixed: The window grows. A system that handles 3000 requests per second is given a little more, and while requests succeed the window keeps increasing. The window shrinks. When requests start failing, the window size is reduced. With 200 requests, the 200th waits a long time. With a window of five, the sixth request is bounced back. A static limit of 200 does not adapt. When the hardware improves, the window sets itself with no code change. A very large window does not help either. A user who waits three seconds hits refresh, and memcached gets two requests on 95 keys.

A lease is an ID that Memcached hands out with every cache miss for a key, so that a late set carrying an old value fails. Memcached starts with zero data. Two web servers miss on the same key, A: The slow web server reads A from the database and gets 10. The fast web server sets A to 20 in the database. The fast web server reads A, gets 20 and sets it in the cache. The slow web server sets A to 10 in the cache. The cache holds 10 and the database holds 20. With leases, the fast web server picks up a new lease ID. The slow web server finds its lease ID is not valid, reads the value again and sets it. A lease ID is similar to a version number and to optimistic locking. Availability is the higher priority, so there is no strong consistency. Read after write is not strict. In rare cases a read does not show the reader's own update.

A slightly stale value is an old cache entry served to a reader who accepts it, so that a key written all the time is not loaded again for every read. A get that misses makes the application server go to the database, fetch the key and populate the cache. A post is liked constantly, so its key is updated constantly and kicked out of the cache. A reader does not care if the likes go from 3,507 to 3,508. So the client says how old a value it accepts. At 8:03 a like count from the last two minutes is fine. A profile picture is wanted within 30 seconds. A delete does not remove the entry: The application server updates the database and deletes the key from the cache. The cache marks the entry as old and keeps it. A reader who accepts data five minutes old is served the old entry. When the cache is out of space, stale entries are kicked out first. Nobody decides beforehand which data may be stale.

An eviction policy decides which entry a cache kicks out when its memory is used up. A cache tries to answer all requests with the minimum number of database calls. It cannot predict what is asked next, so it picks an entry to kick out: the least recently used, the least frequently used, or FIFO. A single eviction policy does not work at Facebook, because the access patterns of its objects are different. Facebook saw two types of entries: Long lived, evergreen entries. People keep using them from time to time, maybe for a day. Logos, ads and profile pictures are like this. Short lived entries. They are popular for a short period and never used again, like a list of 50 entries run through once. Least recently used is good for the short lived entries and poor for the evergreen ones, which are used from time to time. Kept together, the short lived objects kick out the evergreen ones. So the cache is split into two, and each part gets its own policy.

A request that reads many keys at once is served better by replicas of a cache than by its shards. One home page request fans out into a get for keys 1 to 100. Three cache servers answer it slowly, and there are two ways to scale them: Sharding. The servers take keys 1 to 33, 34 to 66 and 67 to 100. The get is split into three pieces, then six when the shards are cut again. With a thousand users every shard still gets a thousand requests, so each has to scale from 100 to a thousand. The problem is not solved. Replication. Each server holds all keys from 1 to 100. One get goes to a single server, the next to another. Sharding is far better when there is no aggregation, because a request is routed to the right place. A cache that is out of memory evicts keys. A database cannot, so it has to shard or replicate. In the worst case a cache reads from the database, so durability does not matter.

A cascading failure is a collapse in which the load of one failed server brings down the servers that remain. When a server collapsed at Facebook, it took time to find out, and then scripts had to be run to start it again. In 2011 FBAR took the common parts of the fixes engineers wrote for their failed scripts and made them an API. Even with this automation, a failure takes a few minutes to be detected and fixed. In those minutes the load on the other servers increases: Three servers, S1, S2 and S3, each run at 60% capacity behind a load balancer. S1 fails. Its 60 units of work are distributed equally, so S2 and S3 each carry 30 plus 60 out of 100. At 90% the chance of a server failing is high. When S2 fails, all the load hits S3. Facebook's answer is the gutter, a separate set of servers, S4. All of S1's requests are routed to the gutter before S2 can fail.

A gutter is a small set of servers that takes the requests of a failed cache server. With S1, S2 and S3, a collapse of any one redirects its requests to S4, the gutter. Once S1 is back, its requests go back. This one backup reduced client failures by 99%. The gutter is already working, so it is not autoscaling, and it is common to every server. It adds 1% more server capacity, where a backup for each doubles it. The client's part: A get times out once. The client assumes the server has failed and queries the gutter. On a cache miss in the gutter, the client sets the entry. On a cache hit, it does nothing. Entries in the gutter expire sooner, so a profile changed in the gutter and in S1 is not inconsistent for long. The load is not distributed over the other servers, because a single key can account for 20% of the requests to a server. A server that gets two or three of these keys is overloaded.

Mcsqueal reads the binary log of a MySQL database and broadcasts every committed update to the other databases and to the caches that use them. Adding servers the naive way lets every web server connect to every memcache server. Facebook broke its architecture into clusters instead. The web servers of a cluster talk only to its own memcache servers, and each cluster is responsible for the entire data set. Each cluster has its own database, so the databases are not consistent. A web server that broadcast every set operation to all databases would be too expensive. The log does the work: MySQL notes every operation down in its binary log, which it replays after a collapse. One of five databases commits an update to its log. After some time mcsqueal reads the log and broadcasts the update to the other four. One line before each update, mcsqueal has written a delete of that key from memcached. The relevant caches delete the key. A single update reaches every database and cache, which gives reasonably high consistency.

A regional pool is a single cluster that holds a large object few people read, in place of a copy in every cluster. The group Powai chess players may hold ten thousand people, and few people in the US, China or Australia look at it. So the object is kept in one cluster, close to India. The other caches do not have it, and the pool is not replicated across regions. A read from the US is not answered by the US cache. Mcrouter looks at the key and routes the read to the Indian server. The cross continent call is rare, and the single copy saves a lot of memory. A cache failure impacts a small set of users. Three factors, measured against thresholds, decide that an object goes to a regional pool: It is accessed infrequently. It is large. It does not affect many users, so the blast radius is small. The object is then assigned to a single cluster, and mcrouter knows where to send a key by looking at a routing table.

Cold cluster warmup fills a new cluster's cache from a warm cluster's memcached, which reduces the time to heat up from days to hours. A new cache cluster is completely empty. A get request misses, the value is set, and the next get request misses again, so the cache is useless until it is populated. On a miss the cold cluster does not hit the database. It hits a warm cluster's memcached, a cross region call, because something popular in one cluster is likely popular in another. Mcrouter does this based on a config flag, until the new cluster has normal hit rates. This creates a race condition: Initially K is 5. The cold cluster sets K to 10 in the database. A get for K misses in the cold cluster and hits the warm cluster, which returns 5. Mcsqueal invalidates K in the warm cluster. The cold cluster holds 5 and the database holds 10. The database never tells the cold cluster to reset the value, so it stays stale.

A cold cache is a cache whose hit rate is under the threshold at which it is called warm. A user changes his name at a Facebook web server that has a cold cache and a database. A warm cache stands in another cluster. The web server deletes the user's entry in its own cache and sets the new name in the database. The user refreshes the page. The web server goes to mcrouter, which hits the warm cache, because the probability of finding the entry there is high. The warm cache still holds the old name. Till now there is no problem. Deleting from the warm caches as well means deleting from all clusters in the world. One database update hits clusters in India, the US and the UK, and waits for the responses. So only one cluster is hit at a time, letting go of consistency for latency. Mcsqueal corrects the warm side. It broadcasts the update to all databases and to the caches that care about the value, and the entries become consistent again.

Two seconds of probation is a wait on a key that has just been set, so that the cold cache is not filled with its old value. The cold cache holds the old name because it was fetched from the warm cache. Mcsqueal kicks that entry out of the warm cache, but it does not broadcast the binary log to the source cluster. Invalidating everything is not an answer, because every update would then require an invalidation. So every key has a probation timestamp: A set marks the key and stores its timestamp. A get before that time fetches the value, and the mcrouter gives it to the web application server. The value is not stored in the cache while the key is in probation. Two seconds is a magic number. Facebook saw that most requests are managed within one second, and made it two to be doubly sure. The trade off is a cache that stays stale if the invalidate does not come through in two seconds. A key with no expiry time is never expired.

Cross-cluster consistency is keeping the cache of one region in step with a database that is written in another region. Facebook runs clusters of Memcached in several regions, and every region holds all the data. One region holds the master databases and the other regions hold read only replicas. All writes go to the master because the system is read heavy: read queries are 100 times the write queries. A write and a read in India race: A web server in India writes K = 10 to the US database. The same web server deletes K from the Indian cache. Another Indian web server gets a read for K and misses in the cache. It reads K = 5 from the Indian replica, which is the stale value. The MySQL binlog replicates K = 10 to the Indian database. The second web server sets K = 5 in the cache. Every later read finds K = 5 in the cache, so the value never changes.

A remote marker is a Boolean flag that Memcached keeps on a deleted key. While it stands, Memcached does not apply a set for that key. A set updates the database and deletes the entry from the cache. A get that misses reads the database and sets the cache. Across regions this races: A write updates the US database and deletes the key. A reader misses, picks the old value up from its database and sets it in the cache. The cache is stale till the next update. With a remote marker, the delete tells Memcached that a remote update is coming. Memcached takes a set, gives a success and does not update the data. The system that reads the MySQL bin logs updates the Indian database and the cache. There is only one remote marker and no counter. After two writes, a reader that comes once the marker is removed can still leave stale data. The cache does not keep the updated value. It evicts it, which makes the rest of the system simpler.

Adaptive memory slabs are the sizes of memory a memcached server stores its objects in, and the number of blocks it assigns to each size. Objects of one type are of similar size. Memory slabs are normally given as 1 KB, 2 KB, 4 KB, 8 KB, and those jumps are large. A small object in a large slab wastes the rest of it. Facebook defines its own slabs, starting from 64 KB: 72, 76, 80, 84, 92, up to 128 and on to 1 MB. The factor is one point zero seven instead of two. Memcached sees which objects are accessed. If profiles are accessed a lot and groups are not, it evicts the least recently used group and breaks that slab into pieces of 64 KB. The web server and mcrouter have no idea. A cache can always have a miss. Short lived items go in a circular buffer of linked lists. Every second the current position moves forward and that linked list is deleted.

TAO is Facebook's graph store, a wrapper of graph APIs over memcached, backed by MySQL. Objects are nodes and associations are edges. The store takes 1 billion reads per second on petabytes of data, and writes are few. The API adds an association, gets the associations of a source and a type, and counts them. Each cache shard holds a range of keys: Routing. The client hashes an object ID to find its shard. An edge goes by the hash of its source ID. The inverse edge. A like has an inverse, liked by, whose source is the post. One like hits two shards. The order. The two edges are written one after the other, so a failure can leave one hanging. A cron job finds the hanging edges. Writes and reads. The writes of a key range go to one place, and many replicas serve reads. Read after write. A write goes to the nearest replica, which forwards it.

TAO is Facebook's in-memory graph based cache with a minimal API. In 2009 Facebook was answering a lot of graph queries. A profile page fetches the friends, the liked posts and the places of one person. Another query asks whether Gaurav is a friend of Anubhav. MySQL and Memcached hold columns and key value pairs, while a graph is a set of objects linked to each other. So each team wrote its own objects and relations in code, and repeated the optimizations of the other teams. TAO rolls them into one library. TAO is a real time store. An analytical query, such as comparing the locations of two people, runs in batch on a data copy. The requirements: 1 billion reads per second. Of every thousand queries, 998 are reads. Multiple petabytes of data, in a graph that is hard to shard. A pull architecture in the clients, which can cause a thundering herd. A read hit rate of 95% and a read latency of 25 milliseconds. Cost is not discussed.

A simple cache like memcached is a key value store, and a graph kept in it loads whole lists and brings concepts irrelevant to graph operations. Edge lists. Every load operation fetches the entire list. Gaurav has almost 5 thousand friends, so showing them brings 5 thousand people into the cache. One edge. To know whether Pankaj liked a post, the post is the key and the whole array of likes is loaded into memory. Concurrent updates. Pankaj likes the post while Sanchit dislikes it, and one array has to allow both. A mismatch needs optimistic locking, pessimistic locking or a merge. Consistency. Memcached ensures it with leases and remote markers. An operation of one or two steps in a graph is three or four in a key value store. TAO is a library over memcached with its own data model of objects and interactions, and it takes care of the cache and the database in the background. There is no real efficiency gain. Engineers think in graph models and not in key value models.

TAO's data model has two records, an object and an association. An object has an ID, a type and a set of key value pairs. A profile with the name Gaurav is an object. An association links two objects. It has a source ID, a type, a destination ID, a timestamp and a map of key value pairs. Gaurav is linked to India by an association of type nationality. The timestamp is there for two reasons: Ordering. One request comes first, another later. Recent event sorting. Most objects that are accessed were created recently, and old posts are rarely seen. The source ID, type and destination ID are unique. A post is liked or not liked, so an association maps a one-time event well. A second check-in at the same hotel is not allowed. Objects map repeatable actions. Each comment on a post is an object, and so is a check-in. An inverse makes an association bidirectional. Gaurav is friends with Pankaj, so Pankaj is also friends with Gaurav. The API adds, deletes and updates an association.

TAO offers four queries on a source and a type of association. Get. It returns the associations from the source to any destination in a set, between a start time and an end time. Count. It returns the count of all edges of the type from the source. Range. It takes a position and a limit, such as the first 50 after an offset of a hundred. This helps pagination. Time range. It takes a start time, an end time and a limit, such as the top 10 posts Gaurav has liked in the last three days. The 50 most recent comments on a post is a time range. The likes on a post are a count. A check-in can be stored in two ways. In the first, Gaurav has an edge to a check-in object, which has an edge to Taj. It is more flexible, but a query goes through two edges. In the second, each check-in object holds the place, so a count of places in the last three days is one query.

TAO is a cache tier in front of a database tier. Client requests are routed to the appropriate cache server, and the cache tier can manage any TAO request. The database tier handles multiple shards. The data has to be stored somewhere, and the database is MySQL, which Facebook has used for a long time. A request that the cache cannot serve moves on: A cache miss. The read goes to the next layer, which is another cache or the database. A write. It goes to the database, because a write needs high consistency. The cache contains commonly queried objects, the likes on each object, and how many people have liked it. Its replacement policy is least recently used, the same as memcached. The cache also gives smart responses. A cached count of zero answers a later get with the same parameters with an empty set. Brad Pitt has zero recent posts, so a get for his recent posts returns nothing. This reduces latency for a common query pattern, where get and count queries come together.

A hanging edge is an inverse edge that exists without its original edge. Objects are sharded by object ID, and edges by their source ID. Gaurav and every like he has made sit in one shard, so a query for his likes goes to one place. This is data locality. The number of logical shards is much larger than the number of databases, which keeps their load almost uniform. A like is two associations, and the problem starts there: User U1 likes post P1. The like goes to the shard that the hash of U1 gives. The inverse, liked by, has P1 as its source, so it goes to a different shard. The source cache calls the remote destination cache, which hits its database and responds. Only then does the source cache write to its database. A failure in between leaves liked by without the like. A two-phase commit is slow, and a quorum has to wait. TAO repairs these edges with a job that runs periodically, so the system is mostly consistent.

TAO writes the edge on the destination before the edge on the source, and fails the request if the destination fails. A like is written in India and in the US. If India persists first and calls the US asynchronously, a failed call leaves a hanging edge, and the remote call is more likely to fail. So the source checks with the destination first: The destination fails. The source fails the query and the client retries. The destination succeeds. The source writes and sends a successful response. The destination succeeds and the source fails. No success response is sent. This case is rare. The other option is to write the source and retry the destination. The client is told everything is okay while a dead database is retried. A cron job is simpler, because it executes once. A smiley changed to angry shows the difference. The source has no like, so the update fails. With retries the like is updated on the source, nothing is on the destination, and every later update is retried.

TAO keeps a copy of its caches and databases in every region, and sends every write to one of them. One deployment in the US makes each like from India a call across continents, so the data is replicated around the globe. A social graph cannot be split by location, because a person in Hyderabad has relatives in the US. A write made in India has to reach the US too. The log. Every create, update and delete is a line in a log. Replaying the lines rebuilds the database, which makes a replica. The exchange. Regions that all take writes exchange logs and replay the lines each is missing. In MySQL this log is the bin log. The cost. When every region is a peer, each database spends its time on updates from the others, and throughput goes down. The leader. Of a thousand operations, two are writes. All writes go to one region and every region serves reads. Clients always connect with the closest follower, never directly with the leader.

In TAO a client reads from a follower and writes through a follower. It never talks to the leader. The write takes one path: The client. It tells a follower it has a write operation. The follower. It talks to the leader. The leader. It does the write and comes back to the follower, which tells the client. That follower and the leader have the latest data. A client that goes to another follower sees stale data, until change data capture propagates the write from the leader to all followers. The leader has rate limiting, and the followers ensure it does not get overloaded. Across regions the follower sends the write to the leader region, which updates its database. The MySQL bin log propagates it to the database in another region. Inside a region the caches have one leader. The leader cache serializes the writes, and the database propagates its updates to that one cache alone. 99 point 8 percent of queries are reads, so one leader in each region is enough.

TAO lets go of some consistency for availability and efficiency, and a follower catches up with a write through an invalidation or a refill. The CAP theorem rules out perfect consistency with perfect availability. TAO serves stale data rather than go to a remote database. A post has 300 likes, the latest count is 301, and TAO keeps 300 for now. Writes are sent to the master. The follower that got the original write request gets a response after success, so it is consistent. Other followers are sent invalidation messages on a queue. A follower with an equal or greater version number ignores the message, and one with older data accepts it. The leader serializes concurrent writes, so a larger version number is always newer. An invalidation deletes the entire key. A group of 20 thousand people gains a member and the list is fetched from the database again. A refill names the updated range. The cache keeps the list and fetches only the new users. Refills are idempotent and can be replayed using logs.

TAO's cache uses the memory layout of memcache. A cache of 64 GB is cut into slabs by the size of the objects, with most of the memory, 32 GB, kept for small objects. Each slab behaves like an LRU cache. Counts are stored separately, in count caches. An entry holds a source, a type and the count in a single 32 bit integer. Negative entries are stored too. A count of zero for the likes on a post answers the query for the people who liked it, and the database is not queried. Empty arrays are no longer stored, which saved 20% of the cache space: three thousand servers became two thousand four hundred. In MySQL, objects are stored in a single table of a 64 bit ID and the object data. Edges are stored in one table: The source, the type and the timestamp sit in one column. That column is the only index. All the likes Pradeep made in the last three days are sorted together, and a single query fetches them.

TAO uses consistent hashing within a tier, and a shard can still become a hot spot. Sharding is on the source ID. A user with too many followers, such as Brad Pitt or Mark Zuckerberg, brings all of that load to one shard. Shard cloning. A hot shard is cloned to rebalance load, so reads to it are served by multiple followers in the tier. Update messages are sent to all the followers hosting the shard. Client caching. A client of TAO is a service that uses it as a database, such as a friends service. The service talks to TAO through a library, and the library caches very hot data. Some lists are very long. Brad Pitt has liked a lot of posts and DiCaprio's post has a lot of likes, so the query is expensive from either side. When the edge and the inverse edge both cross a threshold, TAO queries MySQL. Such a list is not cached, because a few of them fill the cache.

TAO keeps serving when a database or a cache fails, because a request has a second place to go. The leader DB is down. A follower is promoted to leader. A follower DB is down. Cache misses are sent to the leader DB. The leader fails. A read miss hits the DB. A write goes to a random leader in the leader tier, which writes to the DB, writes the inverse edge and sends invalidations to followers. A follower fails. Clients have two addresses for each shard, a primary follower and a backup follower. The Indians try Europe before the US. The DB is the source of truth. A write fails only when the DB fails. Refills and invalidations reach the followers through MySQL's binary log, which works like a message queue. When the acknowledgement is not received, the leader queues the event to disk and tries later. The same write may be sent multiple times, so idempotency is expected. The primary and the backup may not be perfectly consistent. Availability is more important.

A DynamoDB table is read through its keys: the partition key is hashed to find the partition the data belongs in, and an optional sort key keeps the data inside that partition sorted. The partition key plus the sort key is the primary key, one unique key. With the post ID as the partition key and the timestamp as the sort key, the posts of the last seven days come from querying all three partitions and aggregating the results. More access patterns are served from the same table: A composite key. User ID plus post ID returns the posts of user six between 60 and 90 quickly. A local secondary index. The same partition key is sorted by the like count, in the same partition, and the data stays strongly consistent. A global index. It has a different partition key and is an entire copy of the table. Amazon limits them to four. They are eventually consistent, changed by the logs streamed from the original partition. Duplication and denormalization are accepted, and joins are not encouraged.

Kafka is a system from LinkedIn that takes messages from producers, stores them on its servers and lets consumers pull them. Producers publish messages to a topic, which is split into partitions. Messages within a partition are ordered. Partitions sit inside brokers, the Kafka servers. A consumer pulls using its offset. At 7 trillion messages per day, recovery has to be automated: Replicas. Every partition has replicas on other servers. Writes to different replicas leave the data inconsistent. A primary. One primary replica takes all writes, and the other replicas pull from it. When it fails, another replica takes its place. The high watermark. A consumer sees only messages that have been replicated. Batching. Producers send and consumers pull messages in batches. Incrementing the offset after processing gives at least once delivery. Incrementing it first gives at most once. In a consumer group, one consumer pulls each partition. Zero copy pushes a message from the file system to the socket, skipping the application cache.

Kafka is a system that accepts events, stores them in a persistent store and distributes them to all interested parties. Around 2011, large scale internet systems like Google, Facebook and LinkedIn had problems consuming so much data and storing it. Kafka was developed by LinkedIn. A celebrity posts on LinkedIn. That single message has to reach many notification servers. Kafka sits in between: The producer sends the message. Kafka says it got the message and takes care of the delivery, at least once or at most once. Kafka keeps the message for a retention period of one or two weeks and retries it. A subscriber that does not accept it in that time loses it. Publishers only publish and consumers only consume. Kafka scales up as the message rates increase. The architecture becomes three parts: producers, distributors and consumers. Kafka was made for the producer consumer architecture, retention of messages on disk, scale (petabytes of data per day in 2015, recently something like 7 trillion messages per day), high performance, simplicity for its clients and reliability.

A message is the unit of data in Kafka, like a row in a database, and the message key decides which partition holds it. A message carries a body, a partition ID and a topic. A topic is like a queue, and a very large topic is split into pieces called partitions. A message reaches its queue in two steps: The topic. The sender names it. The partition. The hash of the message key, modulo the number of partitions, gives the partition ID. The key can be the user ID of a profile, 356, or any data in the event. A custom router can replace this function. Messages are ordered by topic and partition. One partition is processed first in, first out, never in parallel. Debit 500 from user A and credit 500 to user B, sent to the same topic and partition, are processed in that order, so the final state of the consumers is correct. Partitions give scalability, reliability and ordering. A hot topic takes more partitions, and when one partition fails the others continue.

Kafka writes messages in batches: a producer sends up to 50 events in one call, in place of one call and one acknowledgement for each. A consumer batches too. It names its offset and asks for 50 messages. Kafka sends 50, or all it has when there are fewer. Batching trades latency for throughput. Messages wait in memory, so the system is no longer real time, and throughput increases. Three things follow for a batch: A batch goes to one topic and partition. A producer that writes to three queues holds a batch for each, which consumes memory. The client can force a send of fewer than 50. A batch is a single API call, and its messages are added to the queue in order. A client that collapses loses its batch, unless it stored the messages. The producer and the consumer batch at their own rates, because the messages are stored in between. A batch is compressed: the metadata of 50 messages is reduced. Messages can have schemas in Avro, a format like protobuf.

A consumer group makes sure that all the messages of one topic and partition are consumed by the same member of that group. Without a group, two consumers take messages from the same queue. Debit 500 goes to consumer one, which takes 10 milliseconds, and credit 500 goes to consumer two, which takes one millisecond. The credit entry executes before the debit entry. With a very fast consumer, a transaction T2 executes before T1. Inside a group the messages are accepted in order, distributed in order and processed in order. More consumers than topic and partition pairs are of no use. Two consumers outside a group, C1 and C2, each have their own offset in Kafka, so both consume the same messages. Kafka deals only with the offset. The guarantee is managed by the consumer: At most once. The consumer pulls 50 messages and increments its offset before processing them. If it collapses, those 50 are never processed. At least once. The consumer increments its offset only after it has processed the 50 messages.

A broker is a Kafka server. It receives messages from producers, assigns each one an offset and commits it to disk, like a log. A consumer mentions its offset and the broker reads from there. A single broker serves thousands of partitions. When one server goes down, only some partitions of a topic are affected, and the pieces of a queue are consumed in parallel. Each partition is replicated into multiple brokers for redundancy: One copy is the primary, the others are replicas. A replica can be one message behind, so consumers read only from the primary. If the primary fails, a replica that is in sync is promoted: one that pulled all the messages in the last 10 seconds. A replica that is out of sync cannot stand in the election, but it can vote. The replicas are not read. They are for fault tolerance. Retention is set per topic by time or by storage. When it runs out the oldest message is dropped, and only one member of a consumer group is affected.

The controller is one broker of a Kafka cluster, elected as the overall leader, which assigns the primary of every partition. A single broker serves thousands of partitions, and Kafka has a set of brokers. One of them acts as the leader. ZooKeeper does the leader election for Kafka, through ZooKeeper atomic broadcast. The leader monitors the other brokers for failures. Kafka uses ZooKeeper for cluster management: Registration. A broker registers with a unique ID. Notifications. Subscribers for registration and deregistration listen to /brokers/id. Removal. ZooKeeper has a snapshot of which brokers are alive. A broker that crashes does not respond to health checks on time, so ZooKeeper removes it from the cluster and no new messages reach it. The controller chooses and reassigns the primary nodes for partitions. For partition P1 the leader is L1, and for partition P2 the leader is L2. A topic T1 with 1000 partitions and three replicas for each partition is 3000 pieces of queue. Every partition needs a primary, so the controller selects 1000 primaries for this topic.

Kafka sends a message to consumers only after every in-sync replica holds it. A replica is in sync when its copy of the data is no more than 10 seconds behind. With replicas R1, R2 and R3, and R3 out of sync, a message that R2 and R3 hold is not enough. It has to reach R1 and R2. If R2 fails, the consumer reads from R1. Events are immutable, so a message is deleted with a tombstone, a later message with a null in its value part. It marks a logical deletion for the key, and the previous values for that key go when log compaction runs. A like that is taken back is a null. Zero copy removes a step from sending a message to a consumer: The standard process. The broker loads the message from the file system into its application cache, then writes it to I/O. Those are two I/O calls. Zero copy. The file system cache writes directly to the network channel, without intermediate buffers, in a single I/O call.

Apache Spark is a general purpose data analysis system that keeps intermediate results in memory. The paper is from Berkeley. Before it, each system was specialised: Dremel for aggregation queries, MapReduce for batch jobs, Pregel for graph algorithms. Spark puts them under a single umbrella. A MapReduce job runs on cheap commodity hardware that can crash: The restart. A crash restarts the entire process, which is slow. The disk. Intermediate results are stored and used during a failure. That takes disk space. The memory. Spark stores the results in memory. When a mapper fails, the output before it is resent from memory to a new mapper. The stages. No intermediate results are needed after every map job, so the functions run together in a single computer. Spark focuses only on computation, so the cluster manager is pluggable. An operation is assigned to the computer that has its data in memory.

A MapReduce job takes a big data set through many operations, stage by stage, on cheap computers. Apache Spark is from 2009, Berkeley. Before it came MapReduce from Google: mapper jobs and reducer jobs. The example is a recommendation on Amazon from four tables: orders, profiles, products and views. A user saw P1 the day before yesterday and P2 three months ago, and today there is a sale. The job gives each product a probability of purchase. The tables are loaded into different computers. Each is filtered by user name, date and location. The tables are joined. An IP address is mapped into a country. The output is an item wise probability list. The recommendation system takes the top 10. A map operation converts one type of data into another. A reduce operation joins the tables and gets a single number. So much data means things can fail, so the process is managed, and it runs on commodity hardware to stay cheap. Apache Spark is like the open source, in memory version of MapReduce.

Spark runs the operations of a job together in memory, where MapReduce ran them stage by stage with a store and a transfer after every stage. A common operation joins two tables on one key, such as a hundred GB of orders and 10 GB of profiles on the user profile key. In 2009 this was a big problem. The SQL query ran on the production DB for hours, or on a replica with instances spun up for it. Google built MapReduce, and Apache Hadoop is a MapReduce framework: A job is a pipeline of stages. After every stage the results are stored in an intermediate data store and sent to the next node. The transfers take time, and the chance of a job crashing is proportional to the time it takes. Spark puts operation one and operation two, a join and a filter on the same data, in one step. The engineer writes the steps as a script, and the framework finds the nodes. MapReduce runs with less instance memory. Spark trades memory for speed.

Spark recovers from a crashed node by asking the nodes it depended on for their results, which are kept in memory. MapReduce keeps intermediate results, so a failure in between is not a problem. Spark keeps most things in memory, and intermediate results would lose that purpose. The job is a directed acyclic graph of nodes: Node 6 crashes. It depended on nodes 1, 2 and 3. A new node 7 is spun up. Nodes 1, 2 and 3 give it their results from memory. The data node 6 held is lost forever, and node 7 computes it again. With multiple failures this becomes slow. A checkpoint pushes data to a disk, and one after every line of code takes a lot of bandwidth. One checkpoint at the end is a single movement to disk, but data lost at the last line means performing all the operations again. Hadoop made sense for jobs hours long, because its intermediate results save recomputing after machine failures. Spark has taken those benefits over.

Spark is one general framework for running computations, with libraries in place of the specialised MapReduce, ML, graph and interactive query engines. A deployment scales to 10 thousand nodes. A SQL query is optimised before it runs: the filter is pushed down, so the filtering is done before the join. Data is shared across computations in memory, and disk is the last resort: Interactive queries. A changed query runs on the old result in memory, a 100x speedup. Fault tolerance. If node 6 fails, nodes 1, 2 and 3 bring it back as node 7 from memory, in parallel, a 40x speedup. Streaming. Events are collected for 200 milliseconds into a micro batch and sent through the same pipeline. Merging the new result with the older result is left to the engineers. An event pushed to Kafka goes through the pipeline already defined in Spark. The operations are reduce, collect and foreach. Foreach is a for loop, collect gathers into a list, and reduce takes an average. Checkpoints for sorting and reduce operations happen in memory.

A Spark node keeps its data sets, the RDDs, in a least recently used cache in memory. A node, S1, has 10 GB of memory and 100 GB of data to bring in. The data is streamed, and the data set being worked on is kept in memory. S1 runs three jobs, J1, J2 and J3, and the memory cannot hold all their data: The disk. Data that does not fit moves to disk and can be pulled from disk. The order. First in, first out keeps J3, the latest job, and moves the RDDs of J1 to disk. Least recently used gives preference to a more active J1. The level. Data is evicted on the level of the data set, not the job. A cache holds DS1 to DS6, which belong to J1, J2 and J3. When DS7 arrives and the memory has run out, the least recently used data set is kicked out. It could be DS5, DS3 or DS2. Recently used data stays in memory the longest, which reduces the number of disk operations.

Spark sends the next operation to the node that already holds its data in memory. The cluster manager does not do the placing. It keeps nodes alive, adds nodes and removes nodes. Spark has a management library, which sends operations where they belong: A data set goes to node S1 and the computation completes. The result, RDD2, is in the memory of S1. RDD2 goes through operation 2. S1 has RDD2 in memory, so operation 2 goes to S1, and no bandwidth is used to send results to another server. An add operation on DS1 gives DS2. The next operation is mult on DS2. The map says DS2 exists in S1, so mult is assigned to S1. After a crash a new node is kept in the same rack, else the same availability zone, else the same region. An RDD is a read-only partitioned collection of records. Its preferred locations are a list of nodes, which are tagged. Kubernetes or Mesos is the cluster manager, and ZooKeeper is used for leader election.

Dremel is the interactive analytics system Google built so that a product manager can query petabytes of data in SQL and get an answer in tens of seconds. Before Dremel a query was a MapReduce job, written as code by a software engineer. Every new question meant another job, so a decision waited on engineers. A query runs down a tree: The client fires a query at the query manager. The root sends it to the shards whose key range holds the keys, down through sub shards to the leaves. A leaf holds its data on local disk, with the distributed file store behind it. Each parent aggregates the results of its children, up to the root. Three optimizations keep the latency low: A prefetch cache with a 95% hit rate. Approximate answers, where an exact one is not needed. Columnar storage. The average age is a scan of 3 data points, not 9, and similar data stored together compresses.

Dremel is the system inside Google BigQuery that lets product managers and CXOs query petabytes of data interactively. Earlier, a query ran as MapReduce on the Google File System and took minutes, hours or days. A product manager asked an engineer to write the job, waited two or three hours for the answer, and waited one more hour to change a filter. Dremel was built to give five things: Interactive analytics. A filter is changed and the answer comes in tens of seconds. Short response times. Trillions of records are scanned in tens of seconds. SQL queries, in 2010, when NoSQL was at its height. Product managers and data analysts are comfortable with SQL. Columnar storage. A table is stored column after column, so a search on the city reads the city column alone. Fast aggregations, a sum or a count, with near linear scalability. More nodes take a higher load and latency stays flat. Borg inspired Kubernetes, the Google File System inspired Hadoop, and Dremel inspired Apache Impala and Apache Parquet.

Dremel is used by product managers, operations teams, engineers, support teams and analysts, who expect answers in tens of seconds. It is much faster than MapReduce, and the two differ in three more ways: The program. MapReduce runs any kind of program, and it is code. An engineer writes it. An analyst finds it difficult. The data. MapReduce parses unstructured data. Dremel needs structured columnar data. The work. MapReduce is good for joins across tables, such as Google Photos with Google Maps, and a join in code takes any arbitrary condition. Dremel is for aggregation over a column. Dremel depends on Colossus for storage, Jupiter for the network and Borg, the cluster manager. Dremel is not a database in itself. It is an analytics engine. Google keeps storage in Colossus, compute in Borg, configuration in Spanner and the network in Jupiter, and builds new systems on top of these systems. Monarch used Spanner for configuration and Colossus for storage.

Dremel answers a query by scattering it down a tree of servers and gathering the results back up. Dremel starts with Colossus, the storage, a file store. Clients fire queries as API calls to a query manager, which forwards the query to a root. The root sends the query to all relevant shards. A shard that cannot hold an answer is not picked up. Each shard is a parent node to the nodes under it, through many intermediate layers, down to the leaves. The leaves hold data on a local disk, with a copy in Colossus. Each node aggregates the results of its children and sends them back to its parent. A node that was not used returns zero, the identity of a sum. The local storage is three way replicated: the same data is stored in three nodes. If L3 does not respond, the parent fires the query again at another node with the same data, such as L6. The copies also distribute load. Monarch had something similar: a root node, zones and leaves.

A slot is one thread on one machine, and Dremel estimates the execution of every query in slots. The estimate is a guess: how many machines, into the number of threads those machines have, the query is going to use. A query that pulls 50 thousand data points on threads that each pull 1000 a second needs 50 slots. The client does not ask for slots. The load balancer makes the estimate: A pool. The system has 3000 slots. A query of 2000 slots. It goes ahead. Another query of 2000 slots. Only a thousand are sent, and the query waits. It is slow, or it fails and the client retries, and the system is not overloaded. A node holds several slots. A quad core processor has 4 threads and so 4 slots, an octa core has 8 and a single core has 1. Slots are also what the customer pays for. AWS states how many IO operations a database does per second and charges only for that, and Google charges only for slots.

Dremel keeps a prefetch cache that takes 95% of the hits, and it gives an approximate answer when one is requested. A person's next query is not very different from the last one. It filters on the same data, with a little data added and a little removed. A least recently used cache holds that data, so the first query takes a lot of time and the subsequent queries are quick. An approximate answer is an option for count and top K queries: The top 100 of restaurants ranked 0 to 110. The top 10 and the top 20 are exact, and the tail is sampled from ranks 75 to 110. The top 10 queries of yesterday. The first 8 are accurate, and the others are sampled from ranks 8 to 12. The response is faster because it does not scan all table shards. A query that hits 100 shards has a high chance of a shard failing, and its latency spikes. The election commission asks every person, which takes time. The news channels sample.

Dremel stores all values in columnar format, and a column compresses because its values repeat. A table of the numbers 1 to 20 is stored column after column: 1, 6, 11, 16, then 2, 7, 12, 17. A city column holds Mumbai, Mumbai, Delhi, Delhi. At Google scale a value is repeated millions of times across trillions of records. Age runs from 12 to 112, a hundred distinct values, so it is stored as an enum. Compression saves storage. More data is pulled into memory at a time, and it is pulled faster. This is what makes interactive analytics possible. It has costs: A mapping has to be kept. M M D D is stored as 0011. The data has to be ordered. Errors have to be corrected. Compression and decompression both use CPU. Decompression happens on the root. Columnar storage allows queries to be fired on compressed data, so nothing is decompressed before the queries are fired.

Dremel compresses its data, because compressed data gives faster querying and faster retrieval. The data is a JSON document. JSON is a set of key value pairs, and it also has objects. The schema is a document a search engine has picked up: A document has an ID, links and names. Each link is an object with a set of forward links and backward links. A name has a list of languages and a URL string. A language is an object with a string code and a country. These are classes, and inside them are classes, lists and values. Dremel stores this in columns, not as a row. The compression for a column is non zero run length encoding. Values that are the same are clustered together and replaced with a single value. A column of restaurants in Mumbai, Delhi and Chennai is sorted, so each city sits together. Each city gets a value and a frequency: 3 Mumbais, 2 Delhis, 1 Chennai. Six data points store many data points of a column.

Zanzibar is Google's authorization system, the one service that decides who can view, edit or own a document. It stores trillions of access control lists and answers millions of queries per second with five nines of availability. More than 99% of the queries are reads. A permission is a relation written in a single line, and a group can stand where a user does. The relations form a graph, and a check spreads out as a tree. It is built like a multilevel cache: The servers. ACL servers keep the most used lists in memory and answer the reads and the writes. The store. Spanner holds the namespace configurations, the relation tuples and a change log. The shards. The data is sharded by object ID, so every relation of a document sits in one shard. The background. A watch service streams the change log to clients, a job creates snapshots, and Leopard uses both to pre-compute user sets.

Zanzibar is Google's authorization system, one service that decides who can see a photo, a drive link or a document. Hundreds of systems inside Google need authorization, and Google had five goals in mind when building it: A uniform data model and configuration language. If the data model changes, the older systems have to migrate and change their API connections. Causal ordering of actions. Gaurav adds 10 rupees to his account and Zoom deducts 10 rupees from it. The result of the second operation depends on the first one, so the two are ordered. Arpita's account is independent, and a total order would be wasteful. Massive scale. Trillions of access control lists, millions of requests per second and billions of end users. Very low latency. 10 milliseconds is the 95th percentile, so of a hundred people 95 are served within 10 milliseconds. Five nines of availability. The failure rate is the failures over the total, and it is subtracted from hundred percent. Most of the paper talks about the scale.

A trade-off in a real system is chosen from what the product accepts, not from the CAP theorem. Zanzibar cannot let go of consistency, because it needs causal ordering. It cannot let go of availability, because everybody is using it. It cannot let go of partition tolerance, because the system has to be distributed. CAP is for generic systems. What is acceptable in the domain defines how a system is built. A ledger for payments keeps latency, consistency and availability. A person's balance is the complex computation, and it is managed when the day ends. An authorization system may accept a delay in a write request, such as a password change that takes three minutes. Zanzibar is read heavy. Facebook's Memcache was built for 100 read requests to one write request, and Twitter's write heavy copy did not work out. Read heavy is said of the authorization list. A team can keep editing a document, while writes to its authorization list are much lower than the reads.

Zanzibar uses a simple data model and a powerful query language. A relation is written as one line of text, called a tuple: doc:summary-pdf#owner@123. Doc is the namespace, and photo and drive are other namespaces. Summary-pdf is the object ID, owner is the relation and 123 is the user ID. The paper has three relations: owner, editor and viewer. The query language works on sets of users. Set operations. Union, intersection and exclusion. Exclusion removes a set: a photo is visible to the family, except the people who are currently in China. Computed sets. All admins of a group, or all viewers in the engineering group. User sets. The user ID of a tuple can be a group, written as org:engg#member. A new person is added to org engineering once. Every object that gives access to engineers is then accessible to that person. Tuple to userset expands a set into user IDs: 10, 20, 30, 40. These sets can be used in a union or an intersection.

A logical clock is a set of counters, one for each system, that orders related operations without reading the time. Google ensures causal consistency with globally ordered timestamps from Spanner. Local clocks cannot do it. One computer runs slower, another faster, and they fall out of sync. With three systems a logical clock works: System 1 has the first operation. It increments its counter and shares the clock with S2 and S3, which hold 1 0 0. S2 has an operation, and its clock becomes 1 1 0. S3 has an operation concurrently, and its clock becomes 1 0 1. The first operation is proven to happen before the other two, and those two are concurrent. With thousands of servers the array is huge, and every server maintains all of it. Many servers make operations at once, and all of them are concurrent. Only the next state is ordered after them. Most operations stay jumbled, so related operations are not ordered. Logical clocks are accurate with a small number of servers, as in database systems.

A queue for each object keeps related operations in order and leaves unrelated ones in queues of their own. A blocking queue gives total ordering: the object behind cannot be accessed before the first object is completed. A priority queue does not give causal ordering. Gaurav adds 10 rupees, a get from a premium account is served first, and it returns the old balance. With total ordering only one server can pick at a time, the one which picked the head of the queue. So the queues are split: A queue for every object. Abhitej's operations go in one queue and Ash's in another, like topics in Kafka. One server for one document. The document ID is hashed and gives a server ID, so every operation on document 1 falls on S2 and is processed in order of arrival. A cost. If S2 goes down, D4 and D1 are not accessible. Google went for globally ordered timestamps from Spanner instead. An atomic clock is necessary only for external consistency. Locally the clocks are always ordered.

A zookie is the timestamp a caller passes to Zanzibar with a read. Causal ordering means one action follows another. Gaurav adds Praveen to a document, and Praveen's access has to come after the add. In the opposite order Praveen gets a failure. Google Spanner orders actions in one line, using globally synchronized clocks. A caller asks it for the next number. Gaurav gets 30, Anubhav gets 31 and Adeeb gets 32. These are globally ordered timestamps. A read carries its timestamp to Zanzibar. Zanzibar holds user 10 for the users in Google, and its latest update is at 31: Gaurav sends 30. The data was updated at 31, so the data is good and Zanzibar returns 10. Anubhav sends 31. The two timestamps match, and Zanzibar returns the response it has. Adeeb sends 32. Zanzibar has older data than Adeeb is asking for, so the data may be stale for him. Zanzibar fetches the latest value for the key from its database.

Optimistic concurrency control lets two writers change the same group in Zanzibar without a lock: each changes a copy, and the first commit wins. Tuhin adds a member with a zookie of a hundred while Pooja deletes one with a zookie of 50. Zanzibar does not use the zookie to order them. A copy. Each writer copies the user set at version V1 and changes the copy. A commit. Pooja commits first, and the group is at V2. A rejection. Tuhin's change has V1 as its base and V2 is the current version, so Zanzibar rejects it. A retry. Tuhin takes V2, adds member 555 again and commits V3. A lock is the pessimistic choice. An exclusive lock makes every read wait, and the holder can die with it. Zanzibar is highly available and read heavy, so it does not lock. A third writer can commit before the retry, and the retry is rejected again. A copy and a commit take milliseconds, so a change that takes long keeps failing.

A zookie is the minimum freshness of data that a caller accepts. It is not the timestamp of the query. Pooja holds a zookie of 50. Data at least as fresh as 50 is okay for her, and data before 50 is not. Tuhin holds a hundred. A zookie is not the time of the request, so the writes at 50 and a hundred cannot be ordered and executed one by one. Zanzibar keeps on every entry the time it was last updated. A table of four keys shows it: Anubhav comes with 10. Three entries are okay and one is too old. Pooja is okay with two entries. Tuhin is okay with one entry. Entries updated on or after the zookie are served. The stale entries are fetched from the database. In a cache, data updated three hours ago is good for a caller that accepts four hours and too old for one that wants two. A fifth row at timestamp 3 is not fetched, because the query touches only the first four rows.

Apart from read and write, Zanzibar has three calls, watch, check and expand, and it runs background jobs. Watch. Zanzibar publishes events to all subscribers for a namespace. A person who is out of a company loses all access immediately, because the change is an event. The response contains a zookie. Check. It asks whether a person has access to an object, such as edit access to a video on Drive. It contains a user set, and the set can be nested. Expand. It gives the list of user IDs in a set, such as all the engineers at Google. Three jobs run in the background: periodic snapshots in each namespace, periodic deletion of old snapshots, which is garbage collection, and indexing of large and deeply nested sets. A group may be defined by another group, so evaluating or expanding it takes time. A response is wanted in less than 10 milliseconds for 95% of the time, and a large set of 40 thousand engineers is hard to fetch one by one. Such sets are indexed.

Zanzibar's architecture is caches, backups and subscribers around one database, Spanner. Clients send requests to the Zanzibar servers. These are access control list servers that expand, write and read, and they work as an in-memory cache of access control. Their database is Spanner. Three things run behind the servers: A data pipeline picks up events from Spanner, and a batch job makes snapshots. A snapshot is a copy of the database taken every hour, so a failure still leaves a backup. An indexing system, called Leopard, caches the sets that are commonly queried, whether deeply nested or large. It reads the snapshots, so indexing does not affect the production database. A watch service streams the changelog to clients that subscribe to events. Leopard subscribes too, and changes its index as data is updated. An index built on snapshots is acceptable because many requests carry no zookie. Zanzibar gives them the data it believes is fresh enough. A caller who wants the latest data sends a zookie.

Spanner holds the data Zanzibar needs to decide whether a person is allowed an action: the namespace configurations, the relation tuples and the changelog. A changelog row is keyed by a random shard ID, a timestamp and an update ID. A relation tuple is keyed by five parts: The shard ID. It is usually the object ID hashed into a bucket, and the bucket is a database, so everyone on one document is in one place. The object ID and the relation. This order keeps all the people with edit access on one document together. The user and the commit timestamp. For a very large group, such as a YouTube video that a million people can view, the shard ID is computed on object and user, because the read asks whether one user has access. A small object keeps the object ID, so one query goes to one shard. Inside a shard the rows are sorted on a composite key. Hashing reaches the shard in constant time, and a binary search finds the viewers in logarithmic time.

A write goes to Zanzibar and not straight to Spanner because a write here is specific to the application. Spanner is a database. It can manage writes and order them. It has no idea of auth, ACLs or groups, just as it has no idea of orders in an e-commerce system. Zanzibar has the idea. When Abhishek leaves a team, his access is removed in steps: The groups Abhishek is in are found. The documents which belong to those groups are found. These are read operations, an expand and a check. If the reads succeed, the write is done. This is a conditional write, like a mini transaction. If the balance was sufficient, then deduct the balance. The database is too generic for it, so the logic belongs to the server. Logic on the server brings back an old problem. One server deducts a balance and another adds to it, on the same DB. In Zanzibar, one person removes someone from a group and another makes them its chairman. Both cannot be accepted. One has to fail.

A Zanzibar server is a cache over Spanner, and a write to Spanner leaves the cached entry as it is. A middleware that stores its data only in the database is stateless. A Zanzibar server also stores data locally, in a cache, so it holds state. When a write arrives, Spanner is updated. The cache could evict the entry eagerly. The lesson assumes a simple way of operating in which it does not: The write. It flows through to the database. Spanner holds the new entry at 101 and the cache keeps the old entry. The read. A read operation sends a zookie, or names the previous write. The refresh. When a read asks for a value with a greater timestamp than the cached one, the cache updates itself. Write operations never update the cache. Only read operations fetch the latest value, and the freshness value is there for that. Zanzibar is a read heavy system. An order of magnitude is 10 times, and it has a hundred times as many read operations as writes.

A snapshot in Zanzibar is Spanner's changelog run again somewhere else, and replication is the work of Spanner, not of the Zanzibar servers. Snapshots are backups, and the backups are used to index the system. A database keeps a log of every change. An instance that crashes looks at its log, and a failed transaction is undone with more operations. The database can be thought of as its log. Keys 1 to 5 hold the values 10, 10, 10, 20, 20. There are two ways to copy them: Send every key value pair. Send the log (add 10 to 1, add 20 to 4, and so on) and repeat the operations in the same order. The result is the same database. The snapshot is made the second way, from the changelogs in Spanner. The five voting replicas are in the USA, in three regions, and use Paxos. The other replicas have no vote and still get the latest values. Replication happens in Spanner. The Zanzibar servers are caches, and a cache evicts old entries.

The default staleness is the freshness Zanzibar picks for a request that carries no zookie. Replication happens in Spanner, and Paxos does not wait for all nodes. Of five servers, three persist a write and the write is safe. 2PC requires all nodes to agree. So the local Spanner can be behind. A request in US West asks for key 1 at a timestamp of a hundred, and the Spanner there holds only 98. Zanzibar reads from another zone, such as US East. An out of zone read takes time. Zanzibar counts these reads. 10% or 20% of all requests leaving the zone is too many. A key whose reads stay in zone is marked safe. Its average staleness, say 300, becomes the default freshness. A request without a zookie gets that default. A key three years old is too old, and Zanzibar fetches it from the database. Confidence intervals and z scores predict whether a read stays in zone. None of this impacts an operation with a zookie.

A check is the core call of Zanzibar, and it is evaluated as a boolean expression. The set operations of a check are converted to boolean operations. A query for all users with access to a document breaks into three: the engineers of the company, the admins anywhere in the world, and the public who hold a particular key. The branches are computed in parallel. With AND on all three, one branch that fails means the entire expression fails, so the other branches are cancelled. This is a branch cancellation. The same ACL checks are grouped together. With replicas, a read for 1 can hit node X and a read for 13 can hit node Y, although either node holds both. One query for 1 and 13 hits one replica, so Zanzibar groups the checks to minimize the hits on Spanner. A wide group is expensive to expand. The admins may be the admins of operations, finance, engineering and HR. The expansion is cached in an indexing system called Leopard, which reads off snapshots.

Zanzibar is a distributed cache, and a request is served by making calls between servers and chasing pointers. Take one query: is Adeeb a member of Zoom? One Zanzibar node may not be able to answer it, so it hops from one node to another: The members. One node stores who the members in Zoom are, and they are fetched. The concentric relation. Admins are by default members, as editors of a Google doc already have view access, so the admins are added. The tuple set. All the members and admins are fetched and checked for Adeeb. To minimise RPC calls, the data is sharded by object ID. Zoom is the object and Adeeb is the user, so member, guest and admin in Zoom are kept in one shard. The query now costs one call to reach the shard, plus zero for the admins, plus zero for the tuple set. One user's permissions still hit multiple shards. The usual query asks whether a person has access to a document, so the access pattern matches the object ID.

Zanzibar caches the results of hot keys on both the caller and the callee, and collapses identical requests that are in flight into one. A check on the same group repeats its lookups, so the members, the admins and the tuple set are each cached. Cache keys have timestamps in them. The evaluation timestamp, or zookie, is rounded to the next near value, which lets most requests use a cached result. Request collapsing protects the database from a viral key: The first request for a key goes to the database, and the key is stored locally with a dummy value. A later request for that key finds the dummy value and waits. When the first query returns, every waiting caller is answered together. A caller cancels the branches that can no longer change the answer, and the servers delete them from their queues. A second caller collapsed onto a cancelled request is left hanging. So a server keeps a count of the callers interested in a request, and does not cancel while more than one is waiting.

Request hedging sends the same request a second time, so that one slow or failed read does not hold up every caller waiting on it. Request collapsing takes many requests for the same key into a single one, with one future for the key. If that request fails, everyone gets the failure. A request that takes 10 seconds has wasted ten thousand seconds when a thousand requests were waiting on it. So requests are collapsed up to a threshold: Up to 10 users wait on one request. When the 11th user comes in, another entry is created for the same key with a new future. The same database gets two calls with the same key, and either response is accepted. It stays at two calls, because if both fail it is probably a database or a system error. Zanzibar uses request hedging for calls to Leopard and Spanner, not between its own servers. Every region has at least two replicas, so the second request stays in the region. The threshold of delay changes as per traffic patterns.

A viral post is a single popular object that may have thousands of checks by different users. Zanzibar is indexed on object ID, so a viral post creates a hot spot. Zanzibar detects popular objects by tracking the reads on an object. If the reads are greater than a threshold, all the relations of this object are cached. When the next user comes, his relation is already cached. Rate limiting works in three places: Each client has a maximum of CPU seconds to be used per second. If the CPU seconds are greater than the threshold, as when Photos asks for a lot of relations, the client is rate limited. Each server has the same logic for outstanding RPCs. Requests to Spanner are rate limited for every client and every object and client pair. Different lock tables are used for different clients, so Spanner does not throttle them all. Zanzibar holds 2 trillion relation tuples in 30 regions and answers 10 million queries per second, 99% of them reads. Callee caches save 12% of the reads.

Borg is the system at Google that provides compute and memory for processes and manages each one's life cycle from start to finish. Borg finds the data centers, allocates resources and restarts a process that crashes. 98% of the servers at Google run because of it. A job's priority is monitoring, production, batch or free. A higher priority job preempts a lower one. Production jobs cannot be preempted, which avoids a cascade. A job runs in a Linux container, not a virtual machine. A chroot jail manages security and a cgroup manages resources. A cell is about ten thousand machines under one Borg master, which places a task by a hybrid of best fit and quick assignment. A running task gets a mapping in BNS, the Borg naming service. Chubby stores it in a file with its state and health. The Borg master has five in-memory replicas with re-election using Paxos. If it dies, the Borglets continue functioning. Sending a task where its libraries are installed reduces the start time from 25 seconds to five.

Dapper is the request tracing system at Google: it records the path one request takes from the client to the server and every data store, and how much time it took. A tracing system that slows production is not adopted. Dapper has to be low latency and used by all services, so it is added to the generic client that makes every remote procedure call. A span. Each remote procedure call is a span with an ID, a parent ID and an array of events. A trace. All the spans of one request carry the same trace ID. Collection. Clients write log files to local storage. A background process buffers them. Collectors pull them, convert them into spans and store them in Bigtable, which has an index on the trace ID. Most requests reach Bigtable within 15 seconds. Sampling. IO calls are the bulk of the latency, so production samples one request in a thousand and twenty four. Adaptive sampling. A service with little volume asks for at least 10 samples per second.

Monarch is Google's in-memory time series data store for server side metrics. It takes in 6 million data points per second, holds petabytes of data and answers in less than a hundred milliseconds. It has no hard dependency: if Spanner or Colossus goes down, Monarch keeps running. It replaces Borgmon, which each team deployed on its own servers: Petabytes read at any time are kept in memory, which is very expensive. A histogram is stored for every time window, with an example request for each of its windows. Shared timestamps save 90% of timestamp storage, and differential encoding compresses every time series by around 90%. Leaves hold the data, under 30 or 40 zones and one root. A query goes root, zone, leaf, and results are aggregated on the way back. A field hints index of hashed trigrams skips leaves that cannot hold a key. Queries execute 1000 times faster. Data is stored in three leaves. A soft timeout and hedged requests keep a query from waiting on a slow zone or leaf.

Scalability Low Latency High Availability

A data store is designed from its access patterns: how its data is read and how it is written. A spacecraft has to store data on very limited power. MySQL takes a 30 volt battery and the requirement is 18 volts, so a new database is built. The goal is to do less work, or to do the work in a way that is natural to the system. Memcache and TAO are read heavy: 99.9% of their queries are reads and 0.1% are writes. Writes are so rare that only one master takes them, in the US, and many read replicas sit around the world. A time series database has a lot of writes and very few reads, so that design does not work for it. Two more things are specific to time series: Aggregations. The number of customers who purchased a product hourly, monthly and yearly is a very common query, so the aggregations are computed before the query arrives. Timestamps. They are stored in a compressed format, which saves storage.

Gorilla is the time series database Facebook built. In 2013 Facebook kept its time series data in HBase. A query from an engineer or an external system took a few seconds. Automations timed out, and the time a query took was erratic. Engineers stopped using the system. For a query such as new friends added in the last seven days, they cached the data in their own stores. Many engineers worked on the same problem. The effort was duplicated, and engineers were the most expensive resource. Facebook decided to build one general time series database for the queries of all engineers. Engineers adopt a system only when it is easy to use and easy to understand. The existing system was 2 petabytes, and migrating it is a challenge. OpenTSDB, Whisper and InfluxDB are more flexible, but they are on disk, and a response takes seconds. Memcache is well tested at Facebook and is great for read-heavy systems. In some of Twitter's caches 60% of the queries are writes, and Memcache performs poorly there.

Gorilla is an in-memory time series database that Facebook uses for monitoring. It has four requirements: Fault tolerance. If one of the servers of Gorilla fails, writes continue. High availability. All writes go to multiple regions, so some data is eventually consistent. Anomaly detection, for a network outage, a config change or a deployment. A write heavy load. Writes are 100 times more than reads. In Memcached, reads are a thousand times more than writes. Memory is expensive, so Facebook stores only the last 26 hours in memory. 85% of all queries fall in that range. Older data stays in the archive, in a cheaper file system. Small amounts of data may be lost. High availability does not mean durability, and that is not okay for a trading system. Data points are expensive, so each one is compressed 12 times from its 16 bytes. Gorilla stores 2 billion unique time series at a 15 second granularity. Read latency is less than 1 millisecond, because other systems read a monitoring system to fix a failure automatically.

Gorilla is an in-memory cache that stores time series data. A bunch of servers run Gorilla clients. The client is a library. A team imports it and calls its functions. One function writes and another aggregates, which is a read. Three kinds of systems query the data the clients send: Alarm systems. A metric went down by 50% and stayed down for the past three minutes, so an alarm is thrown. Root cause analysis systems. Sales went down because visits went down, and those visits came from people in the age group of 40 to 60. Dashboards, a more manual way of looking at the data. The data is eventually persisted to a file store. It is long-term durable storage, much cheaper than memory and slow to query. A time series is a key and a set of points. The key is the name of the time series, such as server usage. Each point is a timestamp and a value. The data is sharded on the key, so all its points are in the same shard.

Gorilla stores a timestamp as a difference of differences, because the points of a time series arrive at roughly equal intervals. A time series is one key, such as sales, and a series of points, each a timestamp and a value. Gorilla shards on the key, so a series is kept on a single machine, and it compresses both the timestamp and the value. A server emits a point every five seconds, so the timestamps are roughly equidistant: The timestamps. The seconds are 10, 15, 21, 24, 29, 34 and 39. The differences. They are 5, 6, 3, 5, 5, 5. The differences of the differences. They are 5, 1, minus 3, 2, 0, 0. When every difference is 5, all are 0. 96% of timestamps are compressed to one bit, a zero. A value between 32 and minus 32 takes two bits, and between a thousand and minus a thousand, 10 bits. This gives 12x compression on timestamps. Only the first timestamp is stored whole. The rest are differences, or differences of differences.

Gorilla compresses the values of a time series by storing the XOR of each value with the value before it. Most values are close to one another. Server utilization goes from 70 to 60 to 65, or stays at 30. The XOR of two close values is usually small: 35 and 43 give 8. It is not a subtraction, and it is not always small: 16 and 15 give 31. What is stored follows from the XOR: A zero. A zero is stored. Anything else. The value is stored with its leading and trailing zeros. A series that climbs in equal steps, 32, 36, 40, 44, gives XORs that end in the same zeros. The leading and trailing zeros are consistent across the series. Their counts are noted once, 3 and 1, and only the bits between them are stored. With this, 51% of all values are compressed to one bit. The rest fall in small ranges that take a few bits: some take 8, fewer take 10, and the last take 32.

Gorilla keeps its compressed data in blocks of at most two hours, because a query on compressed data decompresses a block from its first value. Decompression is a linear pass. Compression needs less memory and adds latency, so the blocks are kept short. A query finds a series in three steps, all in memory: The key is hashed to a shard. The shard manager knows which node holds that shard. The shard map points to a time series map, and the series there points to its blocks. Each block leads to the block two hours later. These blocks are immutable. Points are appended to a log file, and every 64 KB is flushed to GlusterFS. A flush that fails loses data, and that is accepted. Every two hours the compressed block is persisted. A restarted instance rebuilds its current block from the log files and loads the older blocks. A second cluster runs with no consistency guarantee. When one fails, reads go to the other. After 26 hours without a problem both hold the same data.

A vector is a mathematical representation of an object, chosen so that similar objects get representations that are close to each other. A large language model gets documents, a query and a prompt. All three are context, and the context is converted to a mathematical representation that the model can use. A vector is on the input side of the model. The representation works like a hash: Each object gets a number. An apple is 36, a banana is 56 and a triangle is minus 3073. The fruits are closer to each other. A new object falls near similar ones. An orange is close to 36 and 56. The function is a neural net. The number does not say what is a fruit: a cat may be at 46. A vector has many dimensions. On an animal dimension the banana is at 37 and the cat is far away. The total distance between similar vectors is low, and between dissimilar vectors it is high.

A vector is a 1D array of values for an object, such that similar objects have a low distance and dissimilar objects a high distance. A document was first represented by term frequency: the number of times each term occurs, as a 1D vector. Now a neural network gives the vector. It converts anything into a vector, gibberish included. A word reaches its vector in steps: The token. A word is split into tokens. Eating is split into eat and ing. The ID. Each token has an ID, such as 36 for cat. A number maps to a vector directly. The vector. Each ID is mapped to a 1D array. Its length is fixed for all the tokens, such as 7000 dimensions. Byte pair encoding finds the split. The text is split into characters, and pairs that occur together often are clubbed into one token. So ing and er become tokens. The LLM is a black box. Engineers often store these vectors in a DB.

The values at the second last layer of a classifier serve as the vector of the object. A classifier classifies an object. Millions of food images go in and four outputs come out: fruit, vegetable, cooked food or something else. Inside are layers of neurons, each connected to the next: The last layer. It gives the probabilities of the four classes and spikes the highest to one. The second last layer. The image has been compressed to its useful features, and the numbers here decide the classification. The values of these neurons, put in a one-dimensional array, are the vector. An image of 100 KB becomes an array of numbers. A cat and a dog share features, so their vectors are similar. A chimpanzee is not a pet, so its vector is not. A classifier of pet or wild keeps the vectors in a 1D space. All animals by type need n dimensions. This is the logic of why vectors work. Word2vec was inspired by it, and more efficient algorithms minimize loss in the representation directly.

A vector search finds the vectors in a database that are closest to a search vector. Two vectors are compared by Euclidean distance, dot product or cosine distance. A website has a large number of products. A person describes a product in text: cold, sweet and delivered quickly. The text is converted into a single vector, the search query, and the database holds thousands or millions of vectors. A linear scan. The dot product of the search query with each vector is found, the results are sorted, and the lowest 10 distances give the products to recommend. Netflix does the same with the last 10 movies a person has seen. This brute force scan is slow. Ranges, one dimension at a time. A dessert query is asked how sweet it is, then how cold, down to the closest vector. This is the KD tree. A sort on one dimension. A binary search finds the vector closest in sweetness. Its distance in sweetness is zero and the remaining distances are very large, which always happens with multiple dimensions.

A clustered index groups the vectors into clusters beforehand, so that a search is compared with one representative of each cluster and not with every vector. The clusters are created before any search arrives, as an offline process, because the embedding space does not change. The example has 14 vectors, and a scan of all of them takes 14 comparisons. With clusters the search runs like this: One vector is picked at random from each cluster as the representative of the whole group. The search vector is compared with all the representatives, and the closest one is chosen. Inside that cluster the search vector is compared with the remaining three vectors. The search takes 5 plus 3, 8 comparisons against the original 14. The savings grow as the clusters grow, and K-means clustering makes the clusters. A cluster of 50 thousand points is split into 50 clusters of a thousand points each, and those into clusters of a hundred. This is similar to an address: country, state, pin code, landmark, and in the end the coordinates.

A search that compares only with centroids can enter the wrong cluster while the closest point lies in another one. A cluster can be any shape, and its boundary is a demarcation in the ND space. The mistake happens in this order: An input is compared with the centroids of C1 and C2 only. Comparing with the entire cluster is an order N operation. C1's centroid is close and C2's is far, so the search goes to C1. The nearest point in C1 is 100 meters away. A point in C2 is just 50 meters away, because many points elsewhere in C2 affect its centroid. Vectors that come in later are assigned the same way. The clustered index is fast, and its only problem is accuracy. The vector database builds it in the background and uses it at search time. A relational database searches by sorting in log n. With K clusters there are K decisions at each level, K log n in all. One layer of many clusters avoids the hops at a small scale.

A clustered index answers a search query with the closest 10 results, and it finds them by brute force inside one cluster. Netflix knows the last 10 movies a user saw and has to find the top 10 related movies. A food app has the last 10 meals. These are the search queries. One query is answered in two steps: The cluster. The DB is split into clusters, and the query goes into the nearest one, found through its representation. The brute force. Every point of the leaf cluster is compared with the query, and the closest 10 are the response. Ten queries become the input in two ways: One vector. The ten vectors are averaged, or one representation is chosen among their clusters. This is approximate. Ten parallel queries. They return a hundred results. The duplicates are removed and the top 10 with the lowest distance are kept. A horror movie and a romcom are not similar. Their average is neither, so running the queries in parallel helps.