Counting at scale: top-K, trending, leaderboards

Counting looks easy: the views of a video, the top 10 searches, the different people who watched a match, the ranks in a game. At a small size, one database column does it. At the size of a big app, the simple way breaks.

The interviewer asks: "A video gets 200,000 views per second. How do you count them?" A weak answer says "add 1 to a row in the database". It does not see that one row, or one key, sits on one server. A strong answer spreads the counter, and says when an estimate with a known error is good enough. It also says which questions need an exact answer.

At SDE-2 level you should spread a hot counter and use a sorted set for a ranking. At SDE-3 level you should also use an estimate on purpose, state its error, and merge results from many shards.

A stadium has 50 gates. Each gate has its own click counter, so nobody waits to press one button. At the end, the staff add the 50 numbers.

Why one counter fails

In lesson 37, INCR adds 1 in one atomic step. That is correct, but one key lives on one server (lesson 43). If one video is hot, all its increments go to one server, and the other servers do nothing. A database row has the same problem: each update locks the row (lesson 22), so the updates run one after another.

Sharded counter

Split one counter into K sub-counters, named like views:42:0 to views:42:31. Each increment picks one at random. A hash of the name places the sub-counters on different servers. To read the count, add all K. Two costs follow:

• A read needs K reads. Keep the sum in a cache (lesson 31) for one second, so the K reads happen once a second.
• The shown number can be up to one second old.

A second way is local batching. Each app server counts in memory and, once a second, adds its total to the key with one INCRBY (add a given number). The cost: a crashed app server loses up to one second of its counts.

Top-K

The top-K is the K items with the highest counts, for example the 10 most searched phrases. The exact way is a sorted set (lesson 37). ZINCRBY adds a search to a phrase, and ZRANGE key 0 9 REV returns the top 10. It needs one entry for every different phrase. With hundreds of millions of phrases, one server cannot hold them.

Split by hash. Put each phrase on one of S shards by a hash of the phrase (lesson 43). Each shard keeps exact counts and returns its own top K. A phrase that is in the global top K is also in the top K of its own shard. So merge the S lists and keep the K best. The answer is exact.

Count-min sketch. A sketch is a small table that gives an estimate and keeps no list of items. A count-min sketch has d rows and w columns of counters, all zero. It has d hash functions, one for each row.

• To add an item, hash it in each row to one column and add 1 to that counter.
• To ask for the count of an item, read its counter in each row and take the smallest.

Different items can share a counter, and that only adds extra counts. So every counter is at least the true count. The smallest one is the closest. The sketch never counts too low. The extra amount has a limit. Choose an error setting ε (epsilon) and a failure chance δ (delta). Then w = e / ε, rounded up (e is 2.718). And d = ln(1 / δ), rounded up (ln is the natural logarithm, the ln key on a calculator). With chance 1 - δ, the estimate is at most the true count + ε x N. Here N is the number of all events. The memory is d x w counters, however many different items arrive.

To find the top K, keep a short list of the best K items. For each event, update the sketch and read the estimate. If it beats the smallest item in the list, put the item in. The list holds the names. A sketch finds heavy hitters: items with counts far above ε x N. It cannot rank rare items, because the error can be larger than their count.

Counting different people: HyperLogLog

The number of different items in a stream is its cardinality. Unique viewers are an example. An exact count keeps a set of all the ids, and the set grows. HyperLogLog (HLL) estimates the count with a fixed small memory.

The idea is coin tossing. Hash each id to random bits. The chance that a hash starts with k zeros is 1 in 2 to the power of k, like k heads in a row. If the longest run you saw is k zeros, you have probably seen about 2 to the power of k different ids. One run is noisy. So the first bits of the hash choose one of m registers. Each register keeps its longest run, and a careful average over all registers gives the estimate. The standard error is about 1.04 / the square root of m.

Redis offers PFADD (add an id), PFCOUNT (the estimate) and PFMERGE (join two HLLs). It uses 16,384 registers, at most 12 KB, and has a standard error of 0.81%. Merging keeps the larger value in each register, so the merged HLL counts the union. You cannot add the unique counts of two days, because one person can watch on both days. The cost: it gives only a number, and it cannot list the people.

Trending

A total favours old, popular items. Trending means rising now. Two ways:

1. Time buckets. Count for each minute. The score for the last hour is the sum of the last 60 buckets.
2. Decay. A count loses half of its value every half-life, for example 30 minutes. So the new score is the old score x 0.5 to the power of (minutes passed / half-life), plus 1 for each new event. An item that was busy long ago fades.

Leaderboards

A leaderboard is a sorted set of players by score (lesson 37). One sorted set holds millions of players, and each update takes about log N steps. When the memory or the update rate is too big, split the players over S shards by a hash of the player id. The top K is the merge of the S local lists, as above. The exact rank of one player is 1 plus the players with a higher score. Count them on each shard and add the S counts.

A video app shows the final match of a cricket league.

Assumptions (made up to teach with, not real traffic):

• 200,000 view events per second at peak. One counter key takes at most 10,000 increments per second. We plan for 70%: 7,000.
• The match page is read 5,000 times per second. There are 200 app servers.
• 40,000,000 different people watch the match on day one. On day two, 38,000,000 people watch, and 30,000,000 of them also watched on day one.
• 2,000,000,000 searches in a day, with 400,000,000 different phrases. A sorted-set entry takes 100 bytes. A server has 64 GB, and we use at most half of it for data (our choice): 32 GB. A sketch counter takes 4 bytes.
• A leaderboard has 20,000,000 players and 50,000 score updates per second.

Step 1: the view counter. 200,000 / 10,000 = 20 sub-counters would be at the limit. At 70%: 200,000 / 7,000 = 28.6, so we choose 32. Each takes 200,000 / 32 = 6,250 per second, which is 62.5% of its limit. This answers the predict box: at least 20. Reads: without a cache, 5,000 x 32 = 160,000 sub-counter reads per second. With a one-second cache, 32 reads per second. With local batching, each of 200 servers adds once a second: 200 increments per second on one key, which is 2% of the limit. A crashed server loses up to 200,000 / 200 = 1,000 counts.

Step 2: different viewers. An exact set of 40,000,000 ids of 8 bytes needs at least 320 MB. An HLL needs 12 KB: 320,000,000 / 12,288 = 26,042 times less. The standard error is 0.81% x 40,000,000 = 324,000 people. The true number of different people is 40 + 38 - 30 = 48 million. Adding the two counts gives 78 million, which is about 62.5% too high. PFMERGE gives about 48 million.

Step 3: the top 10 searches. The exact sorted set needs 400,000,000 x 100 bytes = 40 GB. That is more than the 32 GB we allow on one server. Split into 4 shards by phrase: 10 GB each. Each shard returns its top 10. We merge 40 candidates and keep 10.

For a sketch, choose ε = 0.0001 and δ = 0.01. Then w = 2.718 / 0.0001 = 27,183 (rounded up). And d = ln(100) = 4.6, which rounds up to 5. The counters: 27,183 x 5 = 135,915. At 4 bytes each, that is 135,915 x 4 = 543,660 bytes, about 544 KB. The extra count is at most 0.0001 x 2,000,000,000 = 200,000, with 99% chance. Assume the 10th phrase has 14,000,000 searches (0.7% of all). The error is at most 200,000 / 14,000,000 = 1.4% of it.

We checked this with a short program: 10,000,000 searches over 1,000,000 phrases, the same ε and δ. No estimate was too low. The largest extra count was 365, under the limit of 0.0001 x 10,000,000 = 1,000. The top 10 phrases were right. The 10th had a true count of 69,466 and an estimate of 69,550.

Step 4: trending. Phrase A had 1,000 searches 60 minutes ago and none since. With a half-life of 30 minutes, its score is 1,000 x 0.5 to the power of (60 / 30) = 1,000 x 0.25 = 250. Phrase B had 300 searches one minute ago: 300 x 0.5 to the power of (1 / 30) = 293. B is trending, although A has the larger total.

Step 5: the leaderboard. One sorted set of 20,000,000 players needs 20,000,000 x 100 bytes = 2 GB, which fits. But 50,000 updates per second is above 10,000 for one server. At 70%: 50,000 / 7,000 = 7.1, so 8 shards. Each takes 6,250 per second. The top 10 merges 8 x 10 = 80 candidates. A player's exact rank needs the player's score from its own shard. Then each of the 8 shards counts the players with a higher score.

• A sharded counter solves a hot key: 6,250 per sub-counter against 200,000 on one. It costs K reads for a read and a number that is up to a second old. I would try local batching first.
• A sketch solves memory: 544 KB against 40 GB. It costs an error of up to ε x N for every item, and no list of items. I would change to exact shards when the rare items matter.
• HyperLogLog solves the memory of unique counts: 12 KB against 320 MB. It costs an error of 0.81%, and it cannot list people. I would use an exact set when the set is small or must be exact.

"A bigger server fixes a hot counter." One key lives on one server. Spread the counter over many keys.

"A sketch is wrong by a small percent of each count." The error is a fixed amount for all items, up to ε x N. In Step 3 it was up to 200,000. A phrase with 400 searches could show 200,400.

"The unique viewers of two days are the sum of the two daily counts." The same person can watch on both days. In Step 2 the sum was 78 million and the true number was 48 million. Merge the sets or the HLLs.

1. How does a count-min sketch add an item and read its count? Why can it never count too low?
2. A counter key takes at most 8,000 increments per second. The peak is 120,000 events per second, and you plan for 70%. How many sub-counters do you need, and how many do you choose if you want a power of two?
3. A sketch has ε = 0.001 and δ = 0.01. A seller sees "your dish was searched 3,100 times" from the sketch, after 3,000,000 searches. The dish was searched 400 times. What went wrong, and what do you change?