Loading...
Estimate the scale of the system. Consider daily active users, read/write ratio, storage requirements, bandwidth, and any relevant QPS calculations...
Assuming 1 billion DAU, each user on average uploads 1 photo/video a day, and view 10 posts per day. And assume peak QPS is double of average QPS.
So peak read QPS is 1 billion * 10 / 3600 / 24 * 2 = 231K
Peak write QPS is 1 billion * 1 / 3600 / 24 = 23K
For storage, every day we will have 1 billion new posts.
For each post, we will need 1KB to store text, metadata etc., and 5MB on average to store the media.
So on each day, we will need additional 1kb * 1 billion = 1TB of database storage, and 5MB * 1 billion = 5PB of object storage.
As storage increases, we will move posts that hasn't been read for 1 year to cold storage, to reduce costs.
Define the APIs expected from the system. This is your chance to analyze and define the read and write paths so that you can come up with the high-level design...
To upload a video and image:
POST v1/upload_post {
user_id: UUID,
created_at: Timestamp,
media_link: String,
visibility: String.
title: String,
text: String,
tags: String
}
To follow other users:
POST v1/follow_user {
user_id: UUID,
followed_user_id: UUID
}
To unfollow users:
DELETE v1/unfollow_user {
user_id: UUID,
unfollowed_user_id: UUID
}
To view home page feed:
GET v1/get_home_page_feed {
user_id: UUID,
current_time: Timestamp,
geo_location: String,
cursor: String
}
To send a notification to user:
POST v1/send_notification {
user_id: UUID,
update_user_id: UUID,
post_id: UUID,
updated_at: Timestamp
}
Describe the overall system architecture. Identify the main components needed to solve the problem end-to-end. Use the diagramming tool to create a block diagram.
Let's walk through all the critical paths. First of all, all requests to the system go through API gateway and load balancers. API gateway does rate limiting and authentications, preventing unauthenticated requests or a single user from sending too many requests. Load balancer uses consistent hashing of user_ids to distribute requests across different servers of a service.
When user A follows another user B, we send a request to the follower service, which makes a dual write to the following of user A and followers of user B.
When user A makes a post, the below actions occur:
When user A tries to read their home page feed, they query the feed service. Feed service first queries redis cluster, to see whether the feed content is available. If not, it queries the underlying cassandra database for get the precomputed feed content for user A. And once retrieving results from cassandra, we SET results in Redis with a TTL before returning, so subsequent retrieval of feed content is fast.
If user A follows celebrities, we inject the latest posts for celebrities into feed content of user A before returning. We retrieve the actual post contents from relational database after getting a list of post_ids for the feed.
If feed content is long, we do cursor-based pagination and return parts of home page feed based on the cursor passed in from the request.
Define the data model. Identify the main entities, their attributes, and relationships. Consider the choice of database type (SQL vs NoSQL) and justify your decision based on access patterns...
For database design, we need 5 different storages for instagram.
Object storage:
First, for storing media like images/videos, we need to use object storage like S3. The videos/images uploaded by users are broken down to blocks, and uploaded to S3. They are also cached at local CDNs for faster access.
Graph database:
For recording follow/follower information for billions of users. We will use a graph database. Each user is a node, and each follow/follower relationship between users is an edge.
As we have billions of users, we will need to shard the graph database by consistent hashing of user_id. So for each user, we can easily query a single shard to see all of their followers/following. However, this sharding schema could lead to hot spot problems. A celebrity's following/follower relationships will be stored on a single shard. For celebrities, we will intentionally split their following/follower relationships on multiple shards, to prevent a single shard from being overwhelmed. Combined with the replication approach described below, we should be able to prevent a single shard from being overloaded.
We will need to create replicas for each shard, to handle high write/read throughout and improve reliability in case a shard fails. In terms of replication, when we have W (number of replicas that need to sync a write) + R (number of replicas that need to sync a read) > N (total number of replicas), we have strong consistency. However, in this case we don't need strong consistency and we want to prioritize write/read latency. So as long as 1 replica sync a read/write, we acknowledge success. The tradeoff here is we will have eventual consistency for following/followers relationship.
Redis cache:
For loading home page feed of a user, we cache the results in redis cluster to reduce read latency. The cache key will be the user_id, while the value is the home page feed content.
The redis cluster will serve as a write through cache. While we may have higher write latency due to this, we can ensure consistency between cache and underlying database. Also when cache cluster fails, we can query the database as fallback.
Cassandra:
For storing the actual home page feed under the redis cache layer, cassandra is the right choice. It easily supports huge write throughput. It is eventually consistent but for home page feed it is acceptable. We don't have background jobs writing to it since the cache layer is a write-through cache, the writes happen as they are written to cache.
For the partition key, we will use user_id. So when we query the home page feed for a user, we can easily find it on a cassandra partition. The clustering column is post_created_at, so we can easily fetch back the past n posts.
Also, we store counters like counts or comment counts for a post in cassandra tables as well.
Relational database cluster:
We will store the post metadata in relational database cluster.
We use SQL database because:
Tradeoff versus using cassandra/mongoDB:
Here are sample data models:
table users {
user_id: UUID (sharding_key),
user_name: String,
user_type: String,
user_bio: String,
user_profile_media_link: String,
user_tags: String,
user_created_at: Timestamp,
}
table post {
user_id: UUID (sharding_key)
post_id: UUID,
post_content: String,
post_tags: String,
post_created_at: Timestamp
}
Deep dive into 2-3 key components. Explain how they work, how they scale, discuss tradeoffs, capacity, and any relevant algorithms or data structures.
For database design, we need 5 different storages for instagram.
Object storage:
First, for storing media like images/videos, we need to use object storage like S3. The videos/images uploaded by users are broken down to blocks, and uploaded to S3. They are also cached at local CDNs for faster access.
Graph database:
For recording follow/follower information for billions of users. We will use a graph database. Each user is a node, and each follow/follower relationship between users is an edge.
As we have billions of users, we will need to shard the graph database by consistent hashing of user_id. So for each user, we can easily query a single shard to see all of their followers/following. However, this sharding schema could lead to hot spot problems. A celebrity's following/follower relationships will be stored on a single shard. For celebrities, we will intentionally split their following/follower relationships on multiple shards, to prevent a single shard from being overwhelmed. Combined with the replication approach described below, we should be able to prevent a single shard from being overloaded.
We will need to create replicas for each shard, to handle high write/read throughout and improve reliability in case a shard fails. In terms of replication, when we have W (number of replicas that need to sync a write) + R (number of replicas that need to sync a read) > N (total number of replicas), we have strong consistency. However, in this case we don't need strong consistency and we want to prioritize write/read latency. So as long as 1 replica sync a read/write, we acknowledge success. The tradeoff here is we will have eventual consistency for following/followers relationship.
Redis cache:
For loading home page feed of a user, we cache the results in redis cluster to reduce read latency. The cache key will be the user_id, while the value is the home page feed content.
The redis cluster will serve as a write through cache. While we may have higher write latency due to this, we can ensure consistency between cache and underlying database. Also when cache cluster fails, we can query the database as fallback.
Cassandra:
For storing the actual home page feed under the redis cache layer, cassandra is the right choice. It easily supports huge write throughput. It is eventually consistent but for home page feed it is acceptable. We don't have background jobs writing to it since the cache layer is a write-through cache, the writes happen as they are written to cache.
For the partition key, we will use user_id. So when we query the home page feed for a user, we can easily find it on a cassandra partition. The clustering column is post_created_at, so we can easily fetch back the past n posts.
Also, we store counters like counts or comment counts for a post in cassandra tables as well.
Relational database cluster:
We will store the post metadata in relational database cluster.
We use SQL database because:
Tradeoff versus using cassandra/mongoDB:
Here are sample data models:
table users {
user_id: UUID (sharding_key),
user_name: String,
user_type: String,
user_bio: String,
user_profile_media_link: String,
user_tags: String,
user_created_at: Timestamp,
}
table post {
user_id: UUID (sharding_key)
post_id: UUID,
post_content: String,
post_tags: String,
post_created_at: Timestamp
}
Let's walk through all the critical paths. First of all, all requests to the system go through API gateway and load balancers. API gateway does rate limiting and authentications, preventing unauthenticated requests or a single user from sending too many requests. Load balancer uses consistent hashing of user_ids to distribute requests across different servers of a service.
When user A follows another user B, we send a request to the follower service, which makes a dual write to the following of user A and followers of user B.
When user A makes a post, the below actions occur:
When user A tries to read their home page feed, they query the feed service. Feed service first queries redis cluster, to see whether the feed content is available. If not, it queries the underlying cassandra database for get the precomputed feed content for user A. And once retrieving results from cassandra, we SET results in Redis with a TTL before returning, so subsequent retrieval of feed content is fast.
If user A follows celebrities, we inject the latest posts for celebrities into feed content of user A before returning. We retrieve the actual post contents from relational database after getting a list of post_ids for the feed.
If feed content is long, we do cursor-based pagination and return parts of home page feed based on the cursor passed in from the request.
For counters like post like counts, likes counts etc. We use distributed counters in cassandra. Each counter is stored in several different partitions of cassandra, and each counter has a key like counter_{post_id}_{shard_id}. Background job perodically consolidates them into the final count.