System requirements

Functional:

1. The system can aggregate ads click and query the click count of an ad within a given period (from minute X to minute Y)
2. The system can aggregate ads click and query the top N most clicked ads within a given period (from minute X to minute Y)
3. The system can support queries by different attribute filters, e.g. country, IP, and user id.

Non-Functional:

1. The system need to ensure the data correctness
2. The system should properly handle delays and duplications
3. The system should be resilient to failure
4. The system should return results in less than 2 minutes of latency

Capacity estimation

1. Assume there're 2 million ads and 1 million DAU, and each user clicks an ad per day on average, so the QPS is 10 ^ 9 / 10 ^ 5 = 10,000.
2. Assume the peak is 5 times compared to the normal period, and the peak QPS is 50,000
3. For each record size, we estimate it is 0.1KB, so every day we need to write 0.1KB * 10^9 = 100GB data, and each month the size is 3TB.

API design

1. Query the count of an ad from minute X to minute Y
   1. GET /v1/count
   2. parameter: ad_id, from, to, filter_id
   3. return the count and ad_id
2. Query the top N from minute X to minute Y
   1. GET /v1/top_click
   2. parameter: count, from, to, filter_id
   3. return the list of ads

Database design

We can store 2 types of data. One is the raw data for reconciliation and data processing replay, and the other is the aggregated data for fast queries.

Raw data

It is what we get from input files. Based on the estimation, we have 10,000 QPS, while we don't often read it for calculation, so it is only write-heavy. Considering the size may grow with time, we can select No-SQL database, e.g. Cassandra, to achieve optimized write performance and easy horizontal scalability.

Aggregated data

It stores aggregated and formatted data based on functional requirements.

Aggregated data fields for click count

1. ad_id
2. click_minute
3. count
4. filter_id

Filter configuration fields

1. id
2. details

There're 2 parts: aggregated data and filters.

Aggregated data is downstream of raw data, so it is also write-heavy. In addition, it is for fast query, so it is also read-heavy. We can also consider the same database as raw data to meet read-write performance and scalability requirements.

Filters data is more static and small, so we can use SQL database.

High-level design

1. The log watcher is responsible for fetching the latest input log files and publishing them to the message queue. The aggregator and database writer pull the data from the queue. We build a queue to avoid overload during the peak period.
2. The first database writer writes raw data into the database.
3. The aggregator will generate aggregated data and publish it to the 2nd message queue.
4. The 2nd database writer pulls the data and writes it into the aggregated data database.
5. We also have a recalculation service. In case there is any failure, we can use the raw data to do the recalculation.
6. The reconciliation will compare raw data and aggregated data to ensure data integrity.

Request flows

Explain how the request flows from end to end in your high level design. Also you could draw a sequence diagram using the diagramming tool to enhance your explanation...

Detailed component design

Message Queue

In addition to avoiding overload during peak periods, message queues can also help guarantee data delivery exactly once and avoid duplication.

Aggregator

Time

The aggregator processes data based on the timestamp. We can either select the timestamp from the client or the system timestamp when we receive the event. Client timestamps may have fraud, but it is more accurate. The system timestamps are more reliable, but there will be delay. Data accuracy is more important, so we would choose the client timestamp.

Time Window

As we design, we aggregate data every minute, so we can use a fixed/tumbling window. In addition, because delay cannot be eliminated, it may happen that an event that falls within minute X is received 5 seconds later. To relieve the situation, we can do watermark to extend the time window a bit, like 15 seconds. The downside is this approach will increase the latency, also, it can't resolve the problem entirely if the delay is huge.

Trade offs/Tech choices

Explain any trade offs you have made and why you made certain tech choices...

Failure scenarios/bottlenecks

Duplication

There could be duplications due to server failures. If aggregator successfully processes data from offset x and gets ack from the downstream message queue, but it fails to send ack back to the first message queue. After system recovers, aggregator will reprocess the data and send it to the second message queue.

To resolve this issue, we can add s3 between aggregator and the second message queue to store the offset processed, like the sequence diagram. To be noted, we need a distributed transaction to cover steps 4-6. If any step within it fails, we rollback everything.

Recalculation

If the server is down, aggregated data in memory will be lost. Or if there is any bug, we also need a recalculation from raw data. But it is hard to process from beginning as the data is huge. To resolve this, we can snapshot system status besides data offset we talked about above.

Future improvements

What are some future improvements you would make? How would you mitigate the failure scenario(s) you described above?