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
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.]
The database design is split between raw data and aggregated data storage to support efficient querying and scalability.
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.
The event-logging service process click events from multiple servers in real-time. This service validates and structures each click event, including fields like ad_id, click_timestamp, user_id, ip, and country, before sending it directly to Kafka for downstream processing.
Kafka decouples producers (event-loggers) and consumers (aggregation services), letting each component scale independently and handle high data volumes efficiently.
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 data from the second message queue and writes it to both the Raw Data Database (for unprocessed click events) and the Aggregated Data Database (for click counts and top N ads). This supports fast, real-time queries.
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:
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.
We prioritizes event timestamps over processing timestamps to more accurately reflect when ad clicks occur.
However, using event time can introduce delays when data transmits over networks or moves through queues.
To manage this, the design incorporates watermarking, which extends the aggregation window slightly to include events arriving just after the window’s close. This helps maintain high accuracy while keeping latency low.
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.
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.
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:
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.