Case study: Ad-click aggregation

An advertiser pays each time a person clicks the ad. The advertiser wants two things. First, a live chart of clicks per minute. Second, an exact bill at the end of the month. The interviewer says: "Design a system that counts ad clicks."

A weak answer inserts one database row for each click and runs COUNT. Another weak answer names a log and a tool and stops. A strong answer separates the fast number from the exact number. It puts each click in the minute when it happened, not the minute when it was read. It says what happens to a click that arrives late, and how a restart avoids counting a click twice.

At SDE-2 level you should draw the flow and do the numbers. At SDE-3 level you should also explain late clicks, exactly-once counting and an ad that gets far more clicks than the others.

Think of a cinema with ten ticket counters. Each counter keeps a count sheet with a tick for every ticket, one row for each minute. The manager can look at the sheets at any time and see a live picture. At night the manager also counts the torn ticket stubs. The stub count is the truth, and it includes a bundle that arrived late from one counter. The money is settled from the stubs.

We build the same two things: a live count and a night recount.

What we build

A person sees an ad and taps it. The click server is a small web server that handles the tap. It does three things:

1. It makes an event: click_id, ad_id and the time. The click_id is a unique ID made when the ad was shown (lesson 23). The click server writes the time, because a phone clock can be wrong by minutes (lesson 49).
2. It writes the event to a log (lesson 59) and does not wait for any counting.
3. It sends the person on to the advertiser's page with a redirect (lesson 7).

The requirements:

1. A live count of clicks for each ad for each minute, within seconds.
2. An exact count for the bill: each real click once, late clicks included.
3. Questions: "clicks per minute for the last hour" and "clicks per day this month" for one ad.
4. No click is lost, and no click is counted twice.
5. Raw events are kept for 30 days, so that we can count again.

Fake clicks from bots are removed by rules before counting. That is a separate job, and we do not cover it here.

Stream and batch

A batch job runs now and then. It reads a stored pile of events and writes results. A stream job runs all the time. It reads each new event from the log as it arrives, and it updates the results a little at a time.

We use both. The stream gives the live number. The batch recounts each night from the raw events and gives the exact number for the bill. A second consumer group (lesson 59) copies the log into files in object storage (lesson 24), one set for each hour. The batch reads those files.

Windows

A window is a slice of time that has one count. We use tumbling windows: equal, with no overlap. One window is one minute, for example 10:04:00 to 10:04:59. Each click belongs to exactly one window. A longer range is the sum of its minute rows, as in the time buckets of lesson 82. "The last 5 minutes" is the sum of 5 rows.

Which time puts a click in a window? Event time is the time of the click, written in the event. Processing time is the time when the job reads the event. If the job is behind (lag, lesson 59), processing time puts the click in a later minute than the real one. Use event time.

Late events

A late event arrives after the job has already counted the minute it belongs to. For example, a click server cannot reach the log, keeps the event, and sends it minutes later. A slow network can do the same.

First we need to say when a minute is done. A watermark is the job's statement: "All events before time T have probably arrived." We set T to the newest event time seen minus 2 minutes. The 2 minutes is a grace period, our choice. With many partitions, the watermark is the smallest of their times, because the slowest partition decides.

Clicks that arrive inside the grace period are added to their minute as usual. When the watermark passes the end of a minute, that minute is final: the chart shows it and it never changes. A click that comes after that has two choices. It can change the old row, but then a number that someone has already read changes. Or it can go to a separate late table, saved in the same transaction as the other counts. We choose the late table. The nightly batch counts every click, late ones included, so the bill is exact.

Counting each click once

Repeats come from two places.

First: the job reads the same events twice. A job reads a batch of events from a partition and sums them in memory. It writes the sums, and then it saves its position (the offset, lesson 59). A crash between the write and the save makes the job read the same events again. It counts them twice (at-least-once, lesson 60).

The fix is to keep the position in the same database as the counts, and to change both in one transaction (lesson 21). On restart, the job reads its position from the database and not from the log. Kafka's documentation describes this idea: the consumer stores its offset in the same place as its output.

Second: the same click enters the log twice. The click server retried a write, or the person tapped twice. Both events carry the same click_id. We insert the click_id into a table with a unique constraint (lesson 61), in the same transaction. Only the ids that were new are counted. We keep ids for 10 minutes. A duplicate has the same ad_id, so it goes to the same partition and the same database.

BEGIN;
-- 1. Save the click ids. Only the new ids come back.
INSERT INTO seen_clicks (click_id, seen_at) VALUES (...)
  ON CONFLICT DO NOTHING RETURNING click_id;
-- 2. Add the on-time new clicks to their minute rows.
--    Late new clicks go to the table late_clicks instead.
INSERT INTO live_counts (ad_id, minute, clicks) VALUES (...)
  ON CONFLICT (ad_id, minute)
  DO UPDATE SET clicks = live_counts.clicks + EXCLUDED.clicks;
-- 3. Move the position, only if it is still where we started.
UPDATE offsets SET next_offset = 15000 WHERE part_id = 3 AND next_offset = 5000;
-- If step 3 changed 0 rows, ROLLBACK. Otherwise COMMIT.
COMMIT;

If the job crashes before the commit, nothing is saved, and the restart reads the same events. If it crashes after the commit, the position has moved. Step 3 also stops a second job that wrongly reads the same partition: its update changes 0 rows. This gives exactly-once in effect (lesson 60): the log may send an event twice, but its effect happens once.

One hot ad

The log key is the ad_id, so one job counts all the clicks of one ad. One ad can get a large share of the clicks, for example on the first day of a big sale. Its partition then gets more than one job can handle. This is the hot key of lesson 45.

For ads on a hot list, the key is the ad_id plus a hash of the click_id mod 4 (lesson 43), so the ad usually spreads over 4 partitions. Each job writes a partial row for the ad into its own database. A read asks the 4 databases that hold the ad's rows, which the hot list names, and adds the 4 numbers. This is the sharded counter of lesson 82.

The tables

• live_counts: one row for each ad_id and minute, with the clicks. The stream writes it. There are 16 database servers, one for each partition, so the transaction of a job touches only one server.
• final_counts: one row for each ad_id and day. The nightly batch writes it as a whole value, not as an addition, so a repeat of the batch gives the same result.
• A chart of the last hour reads 60 rows (times 4 for a hot ad). A month total adds the final_counts rows of past days and the live_counts rows of today.

An ad network counts clicks.

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

• 2,000,000,000 clicks a day. The peak is 3 times the average (lesson 14).
• An event is 200 bytes. One job handles 10,000 events per second, and we plan for 70%.
• 300,000,000 different (ad, minute) pairs get clicks in a day. A row is 50 bytes.
• A click is worth Rs 5. 0.1% of clicks arrive after the grace period of 2 minutes.
• One ad gets 12% of the peak clicks. A seen_clicks row is 24 bytes.
• 20 batch workers read 100 MB per second each.

Step 1: rate. 2,000,000,000 / 86,400 = 23,148 per second. The peak is 23,148 x 3 = 69,444 per second.

Step 2: storage. Raw: 2,000,000,000 x 200 bytes = 400 GB a day. The log keeps 7 days: 7 x 400 = 2.8 TB, one copy. Files in object storage for 30 days: 30 x 400 GB = 12 TB. Rows: 300,000,000 x 50 bytes = 15 GB a day. 400 / 15 = 26.7, so the rows are about 27 times smaller than the raw events.

Step 3: jobs and partitions. A job takes 10,000 x 0.7 = 7,000 per second. 69,444 / 7,000 = 9.9, so at least 10 jobs. We choose 16 partitions. Each gets 69,444 / 16 = 4,340 per second, which is 43% of what a job can do.

Step 4: event time. With processing time, the click of 10:04:30 is counted in 10:07, because the job reads it at 10:07:40. With event time it is counted in 10:04. This answers the predict box. With a lag of 3 minutes, every minute's clicks move 3 minutes later. An ad that got 600 clicks in 10:04 shows them in 10:07. The chart is wrong, and an alert on "clicks fell to zero" fires by mistake.

Step 5: a hot ad. The hot ad gets 69,444 x 0.12 = 8,333 per second. The other ads get 69,444 x 0.88 = 61,111 per second, which is 61,111 / 16 = 3,819 for each partition. The partition of the hot ad gets 3,819 + 8,333 = 12,152 per second. A job takes 10,000, so the lag grows by 2,152 per second. After 10 minutes: 2,152 x 600 = 1,291,200 events are behind. A new click waits 1,291,200 / 10,000 = 129 seconds before the job reads it, and the wait grows every second. The live number of this ad is already more than 2 minutes old. The watermark is the smallest time of all partitions, so this one partition also holds back the final minutes of every ad. With 4 sub-keys, the hot ad adds 8,333 / 4 = 2,083 to each of 4 partitions. Each gets 3,819 + 2,083 = 5,902 per second, which is 59% of a job.

Step 6: a crash. A job writes the counts of 10,000 clicks and crashes before it saves its position. Without the shared transaction, the restart counts the same 10,000 again: 10,000 x Rs 5 = Rs 50,000 too much. With one transaction, the counts and the position are saved together or not at all. The extra is Rs 0.

Step 7: duplicate clicks. The table keeps ids for 10 minutes: 69,444 x 600 = 41,666,400 ids at the peak, about 41.7 million. 41,700,000 x 24 bytes = 1,000,800,000 bytes, about 1 GB in total. Over 16 database servers that is about 62 MB each. Each takes about 4,340 inserts per second at the peak. The nightly batch removes any duplicate that came after 10 minutes, by click_id over the whole day.

Step 8: late clicks and the night batch. 0.1% of 2,000,000,000 = 2,000,000 clicks come after the grace period. 2,000,000 x Rs 5 = Rs 10,000,000 a day (1 crore = 10,000,000). These clicks are in the late table, not in the final rows of the chart. If the bill used only the chart numbers, this money would be missed. The batch reads one day: 400 GB = 400,000 MB. The workers read 20 x 100 = 2,000 MB per second. 400,000 / 2,000 = 200 seconds. We assume the whole job, with counting and writing, takes 30 minutes. It writes final_counts after midnight.

• Stream plus batch solves a fast chart and an exact bill. It costs two programs with the same counting rules, which must give the same answer, and the cost of the nightly run. I would use the stream alone for a chart that nobody pays for.
• Event-time windows solve counts in the wrong minute (600 clicks 3 minutes late). They cost a grace period, a late table, and a batch to count the late clicks. I would use processing time only when a small error does not matter.
• Counts and position in one transaction solve double counting after a crash. They cost a database that supports transactions, and one transaction for each batch of events. I would use a unique key for each write when the results go to a store without transactions.
• Sub-keys for a hot ad solve the limit of one partition. They cost 4 reads for a chart, and a hot list to keep. I would use them only for the few ads that need them.

"I count by the time the job reads the click. It is simpler." In Step 4 a lag of 3 minutes moved 600 clicks from 10:04 to 10:07. The chart and the alerts were wrong. Count by the time of the click.

"The log delivers exactly once, so my counts are exact." The log may give the same event again after a crash. The effect is exact only when the counts and the position are saved together. In Step 6 the gap cost Rs 50,000.

"A count-min sketch or HyperLogLog is enough for the bill." They give estimates (lesson 82). A count-min sketch never counts too low, but it can count too high. HyperLogLog can be wrong in either direction. An advertiser pays for each click, so the bill needs exact counts. A sketch fits "which ads are the top 10" on a live chart.

1. What is the difference between event time and processing time? What does a watermark say?
2. A system gets 900,000,000 clicks a day. The peak is 3 times the average. One job handles 8,000 events per second, and you plan for 70%. (a) What are the average and peak rates? (b) How many jobs do you need at the peak? (c) A click is worth Rs 4, and 0.1% arrive after the grace period. How much money is in these late clicks each day?
3. Find the flaw. A job reads 1,000 events, adds their counts to the database, and then saves its position in the log in a separate step. The job crashes between the two steps. What happens, and how do you fix it?