System requirements

Functional:

• aggregate the number of lcicks of ad_id in the last M minutes
• return the top K most clicked ad_id every minitue
• support aggregation filtering by different attributes

Non-Functional:

• data durability
• aggregation correctness
• latency requirement, 1mintue
• roboust

API design

GET /v1/ads/{:ad_id}/aggregated_count

• req
  • from
  • to
  • filter
• resp
  • ad_id
  • count

GET /v1/ads/topk

• req
  • count
  • window
  • filter
• resp
  • list of ad_id

High-level design

• write path:
  • log event watcher watches the event log and sends the events to message
  • event is pulled and aggregated in data aggregation service
  • the raw event and the aggregated event are writtne into DB
• read path:
  • client query the service to get anlytics result
• backup write path:
  • Events in the first MQ will also be written into a DB only for raw data
  • it can be sent to data aggregation service(same binary) in batching for recalcuation for correctness

Deep Dive

• DB choice
  • apparently this is write-heavy system and we choose NoSQL, we can use influx DB, which is time series DB which has a bunch of optimization
• Data aggregation service
  • we uses MapReduce for aggregation in the data aggregation service
  • we use snapshot for failure-recovery
  • We use another message-queue to ensure exactly-once-dilivery
• Time:
  • we use even time instead of processing time for aggregation since(event time is more accurate though it could be un-reliable)
• in the data aggregation service, we does in-memory mapReduce to aggregation
• event can be lagged so the recalculation is necessary
• data can be lagged or data aggregation service could have bug since we need raw-data to recalculate it
• Lambda architecture vs Kappa architecture

Bottleneck, Scalibility and Failure Recovery

Here we shard aggregation nodes by ad_id and there could be hotspot issue. Inside MapReduce system, we need a resource manager to dynamically scale up/down aggregation nodes for this case.

NoSQL can be easily horizontally scaled up with consistent hashing. By mapping virtual nodes on consistent hashing ring to physical nodes.

Future Improvement

How to monitoring system

• end-to-end latency of a metric get generated
• message queue size
• system operation metrics