YouTube Likes Counter: Scalability and Performance

Sharding the two stores

Sharding means splitting a table across many machines, so each machine holds one part of the rows. Both the UserLikes store and the ContentStats store must be sharded to scale out. One database node cannot carry either of them. The real question is which key to split each store on. The two stores need different answers.

UserLikes: shard by user id

We shard UserLikes by user id, or by a hash of it. All reactions by one user then go to the same shard.

This key spreads the write load on its own, because different users act independently. Compare that with sharding by content id. A viral video would then send every one of its writes to one shard. Sharded by user, that video's reactions spread across every shard its reacting users hash to.

Reads by user are cheap for the same reason. A uniqueness check or a toggle needs one user's row for one item. The user id says which shard holds it, so that read is one point lookup.

The drawback is listing every user who liked a video. That would touch many shards. This design never needs that list. The count is kept separately in ContentStats, so nobody ever counts UserLikes rows to serve a page.

In Cassandra, the cluster's partitioner does this spreading for us. The partition key is user_id, hashed so the load is even. Cassandra places the partitions across many nodes and keeps copies of each one. That gives us sharding and failover without extra work.

Two settings still need tuning. The replication factor is how many copies of each row exist. RF=3 in each data center is the usual choice. The consistency level is how many copies must answer before a read or write returns. LOCAL_QUORUM means a majority of the copies in the local region, so two of three. That keeps two local copies in sync while still favoring availability.

One more protection sits alongside sharding: replication across regions. We copy the data to more than one region. Cassandra does this, and so does a globally distributed NewSQL database. If a whole region goes offline, a user's reactions can still be served from another one.

Indexing needs no extra work here. The primary key on (user_id, content_id) is the unique index, and nothing else is needed.

ContentStats: shard by content id

ContentStats splits on a different key. Each item's count is independent of the others, so content id is the natural key. Video ids and comment ids work the same way.

This key carries a risk the user key did not. Some videos are far more popular than the rest. That creates a hot shard, meaning one shard that receives far more updates than the others. It happens when a single video on that shard goes viral. There are ways to reduce the risk.

The first is to use enough shards and a good hash. One very popular item then only has to fit within one shard's capacity, and it usually does. A count update is a light row update, and one database instance handles thousands of those per second. A video taking 50,000 likes per second would need more. Its count could be split across several rows, or held in a distributed counter. But 50,000 per second sustained on one item is extreme. If it happens, give that item its own node or split it by hand.

Some systems use tiered counters. Each region keeps its own count for a video, stored under content id plus region id. A job sums the regional counts into a global total from time to time. That cuts cross-region writes. The cost is that the real-time total gets harder to read, because something has to sum the parts. Eventual consistency is acceptable here, meaning readers may briefly see a total that runs a little behind. So a region could show its local count as a short-term estimate. Still, the simpler choice is one central count per item. One shard can carry 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, meaning a repeat does not give the same result as one. If the write times out and the driver retries it, the increment can be applied twice. Nothing reports that it was. That undoes the guarantee Step 8 exists to provide. So keep ContentStats as 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 ADD update is atomic and safe. But one partition key still caps at roughly 1,000 writes per second. A viral video therefore needs its count split across several keys and summed on read. That is the same repair we describe for the cache below.

The hot key in the cache

Sharding spreads load by moving different keys apart. It cannot split the load on a single key. That limit appears in the cache. Sharding ContentStats spreads the durable writes, but it does nothing for Redis. 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. That is a hot key, meaning one key that receives more traffic than one node can serve.

There are two standard repairs.

Split the key. Store the count as video:abc:0 through video:abc:9. Each write picks one of the ten at random, so the write load divides by ten. A read sums all ten. That costs ten lookups instead of one, but a pipelined Redis call fetches all ten 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. A replica a few milliseconds behind changes nothing a reader would notice.

The two repairs solve different problems. The first 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. Most of it is comment counts rather than video counts. The batch endpoint from Step 4 is what makes that number manageable, and it is worth seeing why.

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 request does one pipelined Redis lookup of twenty-one keys.

The reader's own reaction state is fetched the same way. UserLikes is sharded by user_id, so all twenty-one of one reader's rows sit in one partition. That is one lookup rather than twenty-one.

Notice what this means. 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. So the design has to assume someone will try.

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. That record lets a batch job find what a rate limiter cannot. It sees accounts created together that react together. It sees a video whose rate rises a thousand times in one minute. The correction is an ordinary recount with the flagged rows left out. Step 10 describes that job.

💡 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 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.