0% completed
YouTube Likes Counter: Scalability and Performance
On This Page
Step 9: Scalability and Performance
Sharding the two stores
The hot key in the cache
Serving 250,000 reads per second
Rate limiting and vote manipulation
Step 9: Scalability and Performance
Sharding the two stores
Both the UserLikes store and the ContentStats store must be partitioned to scale out. We cannot rely on a single monolithic database.
UserLikes (Per-User actions) Partitioning:
-
We choose a sharding strategy that balances load and avoids hot spots. A good approach is to shard by user ID (or a hash of user ID). This means all likes by a particular user go to the same partition.
- Rationale: When users add likes, different users are acting independently, so writes are naturally spread across shards by user. This avoids the scenario where one very popular video causes all writes to target one shard (which could happen if we sharded by content ID). By sharding on user, even if one video is extremely popular, the workload is spread across all the users who like it (because their user IDs hash to many different shards).
- Additionally, when reading or updating a specific user's like (which we do for uniqueness checks and toggles), we know exactly which shard to hit (based on user id). This yields efficient point lookups or updates.
- The drawback is if we ever needed to list all users who liked a given video (which we don't in this design), that would require querying many shards. But that operation is not required for serving the like count (the count is maintained separately).
-
In a NoSQL context like Cassandra, the distribution is handled by the cluster's partitioner. We would choose the partition key as user_id (maybe with a hash to ensure even distribution). Cassandra will store partitions across many nodes (with replication). This gives automatic sharding and failover. We would tune the replication factor (e.g., RF=3 in each data center) and consistency level (e.g., LOCAL_QUORUM for reads/writes to ensure the local region has two copies in sync, prioritizing availability).
-
Replication: We replicate data across regions for resilience. For example, with Cassandra or a globally distributed NewSQL database, each user's data might exist in multiple regions. This way if an entire region goes offline, the user's like data can be served from another region.
-
Indexing: The primary key on (user_id, content_id) serves as our unique index.
ContentStats (Counts) Partitioning:
- We shard the ContentStats table by content ID (video or comment ID). Each item's count is independent, so this is a natural key.
- Because some videos are much more popular than others, there is a potential for hot spots: one shard might receive a disproportionate number of updates if a single video on that shard becomes globally viral. To mitigate this:
- Ensure that the number of shards is large enough and use a good hash, so even very popular content's updates can be handled by one shard's resources. If a single counter update is, say, a lightweight row update, one DB instance can handle thousands per second easily. If one video is getting, for example, 50k likes/sec, we may consider more advanced partitioning (like splitting a single content's count across multiple subtables or using a distributed counter). But realistically, 50k/sec sustained on one item is extreme. For planning, if needed, we could assign particularly hot items a dedicated node or do manual splitting.
- Some systems use tiered counters: e.g., maintain per-region counts for a video, then sum them for global. Our design could incorporate that: each region's aggregator updates a local regional count (stored under contentID+regionID). Periodically, those are aggregated to a global total. This reduces cross-region writes but complicates reading the total in real-time (you'd need to sum or have a background job). Given eventual consistency is acceptable, we might simply show region-local counts that approximate global in short term. However, a simpler approach is to centralize the count per item and rely on the ability of one shard to manage it with eventual updates.
- Do not use Cassandra counter columns here. Cassandra has a counter type, and it looks like the obvious fit. It is not. A counter update is not idempotent. If the write times out and the driver retries it, the increment can land twice, and nothing reports that it did. That undoes the guarantee Step 8 exists to provide. Keep
ContentStatsas an ordinary column that the consumer overwrites with a computed total. A replayed batch then writes the same number instead of adding to it. - DynamoDB has the same problem in a different place. Its
ADDupdate is atomic and safe, but one partition key still caps at roughly 1,000 writes per second. A viral video needs its count split across several keys and summed on read, which is the same repair described for the cache below.
The hot key in the cache
Sharding ContentStats spreads the durable writes. It does nothing for the cache, because one video's count is one Redis key, and one key lives on one node. A video taking 50,000 reactions per second sends all of them to that node.
Two standard repairs.
Split the key. Store the count as video:abc:0 through video:abc:9, and have each write pick one of the ten at random. A read sums all ten. The write load divides by ten. The read costs ten lookups instead of one, which a pipelined Redis call fetches in a single round trip. Apply this only to items measured as hot, never to everything.
Read from replicas. A Redis replica serves reads for the same key. Counts are eventually consistent already, so a replica a few milliseconds behind changes nothing a reader would notice.
The first repair is for hot writes and the second is for hot reads. A viral video has both.
Serving 250,000 reads per second
Step 3 put the read load at about 250,000 requests per second, and most of it is comment counts rather than video counts. The batch endpoint from Step 4 is what makes that number manageable.
One video page needs one video count and twenty comment counts. Sent as twenty-one separate requests, that is 250,000 requests per second arriving at the service. Sent as one batched request carrying twenty-one ids, it is about 11,500 requests per second, each doing one pipelined Redis lookup of twenty-one keys.
The reader's own reaction state is fetched the same way. Because UserLikes is sharded by user_id, all twenty-one of one reader's rows sit in one partition, so that is one lookup rather than twenty-one.
The sharding choice above and the batch endpoint in Step 4 are the same decision seen from two sides.
Rate limiting and vote manipulation
A public count is a target. It affects ranking, and moving it is worth money to somebody.
The API gateway does coarse rate limiting per IP address. That stops the crudest scripts and nothing else. Two further layers matter.
Per user limits. A real person produces a few reactions a minute. A limit held against the user id, in the same Redis cluster, stops one stolen account from casting thousands.
Detection after the fact. UserLikes keeps every reaction with a timestamp, so a batch job can find what a rate limiter cannot. It sees accounts created together that react together, or a video whose rate rises a thousand times in one minute. The correction is an ordinary recount with the flagged rows left out, which is a job Step 10 describes.
💡 In the interview: Sharding is where most candidates stop, and it is only half of scaling this system. Two things separate a strong answer. Say that sharding the counts store does nothing for a hot cache key, then give the split key repair. And when abuse comes up, do not name a rate limiter and move on. Say that the per user reaction rows let you find manipulation after the fact and recount without it. The offline half is the part that shows you have run something like this rather than only drawn it.
Next: Step 10, which says what happens when each of these parts fails.
On This Page
Step 9: Scalability and Performance
Sharding the two stores
The hot key in the cache
Serving 250,000 reads per second
Rate limiting and vote manipulation