Loading...
User sees ads while visiting a web site.
User clicks an ad.
Advertisers can learn the following information about the ad:
With 1 billion DAU, we assume each user clicks on one ad daily. This simplification allows us to calculate a base rate of 1 billion clicks per day, which provides a realistic foundation for planning our system's capacity. While actual usage may vary, starting with this average is essential for capacity planning as it represents a standard load for the system.
[Average QPS is a critical measure for determining the system's baseline capacity. It guides the selection and configuration of system components like databases, messaging queues, and application servers.]
To calculate QPS, we consider the entire day, breaking it into seconds:
The average QPS is derived by dividing daily clicks by the seconds in a day:
[Knowing the average QPS helps us avoid costly over-provisioning or inadequate resource allocation.]
User activity is rarely uniform throughout the day. During peak hours, traffic can spike significantly above the average, so we account for potential bursts in load by estimating peak traffic at 5x the average QPS. This results in:
[By planning for peak QPS, we ensure the system can scale up quickly to handle sudden load increases without latency or downtime. This consideration affects the design of auto-scaling policies and informs the choice of scalable databases and messaging systems that can accommodate surges.]
Each ad click event is estimated to occupy 0.1 KB, storing minimal but essential information, such as ad_id, click_timestamp, user_id, IP, and country. With this per-event storage size, we can calculate daily and monthly storage needs:
[This calculation will help us decided which type of database to pick later on.]
[The capacity estimation reveals that the ad click aggregation system needs to handle high throughput with 1 billion daily clicks, averaging 10,000 QPS and peaking at 50,000 QPS. With a daily storage requirement of 100 GB (or 3 TB per month), scalable, cost-effective storage is crucial.
Real-time processing is necessary to ensure low latency for insights, while all components—message queue, aggregation service, and databases—must support high write-throughput and be horizontally scalable.
To maintain billing accuracy, the system requires fault tolerance and exactly-once processing for data integrity.]
We will focus on these two api endpoints since they are part of the functional requirements.
GET /ads/{ad_id}/click_count
Returns the aggregated click count for a specific ad within a specified time range
[This API provides precise click counts for individual ads over specified time ranges.]
GET /ads/popular
Returns the top N ads with the highest number of clicks within the specified time window.
[This API helps advertisers quickly identify top-performing ads within a specific time frame.]
[API design probably isn't the most crucial part of this problem, we should keep it simple as to not spend too much time on this part. They mainly help us identify the read and write paths.]
[Senior level deep dive topics. Database discussion is always a good senior level deep dive topic.]
When considering our storage approach, we have to address both performance and reliability requirements. The main goals are to ensure that our system could handle high-throughput ad click events while providing a reliable backup for recalculating and verifying data as needed. We can achieve this by splitting our storage into two distinct types: raw data storage and aggregated data storage.
[During the interview, it's useful to talk about your thought process like above to drive the interview.]
Raw data is essential for recalculating aggregated data if errors occur. Raw ad click events, including fields like ad_id, click_timestamp, and user_id, are stored for detailed analytics, debugging, and as a backup for recalculating aggregated data. In addition, older data can be moved to cold storage to save costs.
We could store this in a distributed database like Cassandra or in columnar storage formats like Parquet or ORC on Amazon S3. This allows for efficient storage and retrieval of large volumes of data.
Aggregated data storage holds pre-aggregated click counts to support fast real-time queries, reducing reliance on raw data. It includes fields like ad_id, click_minute, and count, with additional fields (e.g., filter_id) for filtering by dimensions like country or IP.
Cassandra or InfluxDB are recommended for their time-series capabilities and scalability, enabling efficient time-based querying.
Distributed databases like Cassandra use virtual nodes to handle large data volumes and evenly distribute the load.
For hotspot mitigation, the system can dynamically allocate additional aggregation nodes to manage popular ads, balancing the load effectively.
To filter and query data efficiently in a NoSQL database, a Star Schema can organize fields like country, user_id, or IP into separate columns or filter_id values, allowing for quick access. This approach is simple, reusable for adding dimensions, and improves query performance by pre-aggregating data along filtering fields.
This system is write-heavy. This means the write path must prioritize handling large volumes of ad click events with scalability, reliability, and minimal latency. To manage this, we will use a message queue pattern that enables processing high-frequency events efficiently.
The first stage in the write path is the Event-Logging Service, which functions as the producer in this setup. This service ingests raw ad click events from multiple servers, validates, and structures them with fields like ad_id, click_timestamp, and user_id. The structured events are then sent to a message queue (Kafka) for further processing. Using Kafka here allows us to manage surges in event volume, as Kafka can absorb high-throughput data without overwhelming downstream services.
[Note: You don't necessarily have to explain every service in depth during the actual interview.]
Kafka decouples producers (event-loggers) and consumers (aggregation services), letting each component scale independently and handle high data volumes efficiently.
Support for exactly-once processing in Kafka using distributed transactions.
This service pulls events from Kafka to perform real-time aggregation, organized in a MapReduce structure with Map, Aggregate, and Reduce nodes. These nodes partition, calculate, and consolidate click counts and top N ads every minute. The aggregated results are sent to a second message queue to ensure exactly-once delivery to downstream consumers.
The second message queue holds the results produced by the Data Aggregation Service, such as minute-level aggregated click counts per ad_id and top N most-clicked ads. This queue ensures exactly-once delivery of these results to downstream consumers, particularly the Database Writer, by buffering data and allowing controlled, reliable ingestion to storage. This approach improves system resilience by decoupling the Data Aggregation Service from storage.
The Database Writer retrieves processed data from the second message queue and raw data from the first message queue, storing each in its respective database. Unprocessed click events are stored in the Raw Data Database, while click counts and top N ads are saved in the Aggregated Data Database. This setup enables fast, real-time querying by separating raw and aggregated data for efficient access and analysis.
This service isolates write responsibilities, allowing the aggregation logic to stay independent of storage, making the system modular. It also enables concurrent writing to both raw and pre-aggregated databases, supporting both detailed analysis and optimized querying.
The Query Service provides a UI or API for advertisers, data scientists, and others to retrieve data. It offers filter options for attributes such as country, IP, or user ID.
Basic request flow should be like this:
[Senior level deep dive topics.]
We can combine real-time streaming for immediate aggregation with optional batch processing for historical data recalculations.
Stream processing handles incoming events in real-time, immediately updating click counts and top-clicked ads each minute.
By contrast, batch processing is reserved for recalculating aggregated data if errors are discovered, enabling retroactive corrections without impacting real-time processing.
This combined architecture, often called the Kappa architecture, avoids the complexity of dual paths (as in Lambda architecture) by using a single stream for both live and historical data.
To calculate the top K ads per minute and click counts, the system employs tumbling windows, which divide time into fixed, non-overlapping intervals (one minute each).
This structure is efficient for minute-level aggregations, but the design faces a trade-off when aggregating over longer windows, like the last 5 or 10 minutes. Storing a larger K per minute (e.g., top 1000 or 2000 ads) enables approximating the top K over broader windows, though this comes with increased storage costs.
This "long tail" approach captures both popular and moderately clicked ads, providing more accuracy across multi-minute aggregations.
Event Time and Watermarking are crucial techniques in data streaming systems to ensure accurate aggregation, especially when handling out-of-order events.
In our system, we prioritize using event time—the actual time when a user clicks an ad—over processing time, which reflects when the server processes the event.
This choice helps ensure that our aggregation accurately reflects real-world activity, even if some events are delayed in reaching the server.
However, relying on event time introduces a challenge. Network latency or queue delays can cause some events to arrive late, meaning they may miss the aggregation window they belong to (e.g., a click that occurred within one minute may arrive a few seconds late). To manage this, we implement watermarking.
A watermark is a threshold that extends the aggregation window slightly, allowing delayed events to still be counted if they arrive within a set period after the window's close.
For instance, if we have a one-minute aggregation window, a 10-second watermark would allow events arriving up to 10 seconds late to be included in the minute they occurred. This approach maintains high accuracy by accounting for network delays while keeping system latency low.
The length of the watermark is a trade-off: a longer watermark improves data accuracy by capturing more delayed events but introduces additional latency, while a shorter watermark reduces latency but may exclude some late events, potentially impacting accuracy. This flexibility helps us balance timely data aggregation with the need for precise metrics in real-time analytics.
Ensuring exactly-once delivery is crucial, given that ad click data directly influences billing. The first message queue (e.g., Kafka) supports this by buffering events and ensuring that each click is processed precisely once.
However, we also include a second message queue to handle aggregated results with the same level of reliability, preventing data loss or duplication in case of failures during storage.
To manage duplicates:
To handle traffic growth, the system can independently scale its message queue, aggregation service, and database components.
The message queues use ad_id for partitioning, which allows multiple consumer groups to process different ad IDs in parallel, with a predefined number of partitions to reduce rebalancing needs.
Aggregation service scaling can be achieved through either multi-threading (allocating ad IDs to threads) or deploying nodes on resource managers like Hadoop YARN.
In the database layer, Cassandra’s horizontal scaling with consistent hashing efficiently distributes data across nodes and automatically rebalances when new nodes are added. High-traffic ad IDs are spread across more nodes to prevent any single node from becoming a hotspot.
The system continuously monitors key metrics such as latency, queue size, and resource usage on aggregation nodes to help maintain optimal performance.
To further ensure data integrity, a batch reconciliation job runs at the end of each day, sorting events by event time and verifying that all events are accurately aggregated.
This reconciliation process helps confirm that any slightly delayed events are accounted for in the final data.
In our design we reuse the aggregation service to handle reconciliation by re-aggregating raw data during the reprocessing phase. This approach allows us to leverage the existing aggregation logic and infrastructure, ensuring consistency in how data is processed, both in real-time and during reconciliation.
[Senior level deep dive topics.]
In our design, we chose Kafka paired with a custom aggregation service for data streaming and aggregation, prioritizing simplicity.
Although effective, this approach requires additional code to manage time windows, state, and fault tolerance, which increases complexity as the system scales. Kafka provides durability and exactly-once delivery, but managing windowing and data consistency demands custom development.
Apache Flink, by contrast, has built-in capabilities for windowing, state management, exactly-once processing, and scalable, high-throughput performance. While Kafka with custom aggregation is more cost-effective for moderate data volumes, Flink excels at handling advanced, large-scale analytics with less custom code, enabling easier horizontal scaling and greater efficiency as data demands grow.
We would consider switching to Apache Flink if our data processing demands increase significantly or if our analytics requirements become more complex.
For faster queries and more complex analytics, we could also use OLAP databases like ClickHouse or Druid in addition to, or instead of, Cassandra.
These OLAP databases are optimized for large-scale data aggregation and retrieval, making them suitable for dashboards and analytical queries that data scientists or product managers might need.
There would also be an OLAP layer (such as ClickHouse or Druid) with Hive as the primary data storage.
This setup is particularly useful for analytics and reporting use cases, offering better query latency and scalability compared to a traditional NoSQL database like Cassandra.
For handling ad click aggregation, multi-threading offers a simpler setup, managing tasks within a single machine’s resources.
This approach is easy to implement but has limited scalability as threads share resources. Multi-processing, on the other hand, uses distributed frameworks like Hadoop YARN to manage tasks across multiple nodes, enabling higher scalability by distributing workloads.
We currently opts for multi-threading to meet demands, prioritizing simplicity and efficiency. However, it can switch to multi-processing if data volume grows, balancing scalability needs with operational simplicity based on current load and projected growth.
[Senior level deep dive topics. Being able to identify potential failure scenario and come up with a reasonable mitigation plan is a good demonstration of senior level engineer.]
Fault tolerance is achieved through periodic snapshots that capture the aggregation state, including offsets and top ads, enabling fast recovery.
If an aggregation node fails, a new node can use the latest snapshot and replay missed events to avoid data loss. This ensures minimal downtime and maintains data accuracy, even in the case of unexpected failures.
Snapshot vs Replay
Snapshots allow the system to resume processing almost immediately by restoring the last saved state, which includes already aggregated data and the last processed event offset. Recovery time is very short as it only needs to replay a small number of missed events after the snapshot.
Without snapshots, the system has to replay all events from the beginning of the aggregation window. This can introduce delays, especially if many events occurred since the last processed point, making recovery slower as the aggregation window grows.
However, if the aggregation window is small (one minute) then replay would still be generally fast.
But, snapshots could still provide value if:
If two threads or nodes are handling updates for the same ad concurrently and don’t manage atomicity, the click count might get updated incorrectly. Suppose one thread reads the current count as 50, another reads it as 50 too, and both add their updates without awareness of each other. The resulting count could be recorded as 51 or 52 when it should be 52.
Some mitigations we could use:
Sharding: Assign each ad (or ad partition) to a single node or thread to prevent multiple threads from updating the same counter simultaneously.
Atomic Counters: For systems that require in-memory counters, use atomic data types to ensure consistent updates to counters.
Distributed Transactions: If the aggregation spans multiple nodes, Kafka transactions or another distributed transaction management system can ensure atomicity.
During peak times, the system could switch to smaller aggregation windows to reduce latency and maintain accuracy, while during low-traffic periods, larger windows could reduce resource usage.