System requirements

Functional:

• Track and store web site user behavior
• Provide website owner data around visitor statistics, conversion rate, audience segmentation
• Provide real-time statistics

Non-Functional:

• System is high available (99.99%)
• Eventual consistency is okay
• Latency of queries should be <500ms
• Real-time statistics latency <1min
• Respect user privacy, only provide aggregated data
• System highly scalable and fault tolerant

Capacity estimation

Assumptions

• 1M websites tracked
• Each website on average gets 1k visitors, where each visitor makes 20 clicks.
• Assume each click need to register 1k of data

Storage

• Daily 1M * 1k * 20 * 1k = 20TB
• In one year data size is 7PB

API design

// log request

logAction(user, url, action, timestamp)

action can be structured type like

{

actionName,

sessionId,

metadata, // {kv pairs}

}

// query path

getData(query) // query is a custom query format

Database design

Since the log data is not strongly structured, there is weak relationship between each row, and we have high write throughput, we choose a no-sql database like Cassandra, DynamoDB.

Schema

• user // anonymized
• url
• action
• timestamp
• metadata

High-level design

IngestService

• Ingests analytic logs from client

LogQueue

• Queues incoming logs for downstream processing

LogProcessingService

• Validates the data
• Anonymizes PII
• Enriches data (e.g. pull in user segmentation dimensions)

AggregationService

• Pre-aggregates data based on user configuration
• Stores result back to database

ReportClient and service

• Supports web analytic user to query data
• Implements dashboards

RetentionService

• Moves distant data to cold storage

Request flows

Ingestion flow

1. Client sends log(s) to IngestService via Gateway
2. IngestService pushes the message onto LogQueue
3. LogProcessingService picks up the log, and performs
   1. Validation: make sure user token is valid, authenticates user (if applicable)
   2. Enrichment: joins other parts other database (e.g. User table) to support segmentation
   3. Anonymization: hide PII data
4. The processed data is written to DB
5. AggregationService based on user configuration, pre-aggregates data (e.g. user clicks every 5 min, 1 hour, 1 day, 30 days), and stores them in database

Query flow

1. The reporting client sends a query to ReportService via another Gateway.
2. ReportServer translates the query to db query, and return the result.
3. For aggregated data, it queries SQLDB which is curated.
4. Optionally we allow ReportService to query raw data. This should be controlled (e.g. have proper timeouts, enforce row limits) so as not to negatively affects other parts of the system.

Detailed component design

Database

• Say we need to store 3 months of data in hot storage, that amounts to 2PB of data. We need some kind of sharding.
• Say each shard should have ~20TB of data, then we need 100 shards.
• We partition the log data by the website. This will ensure relevant data are collocated on the same database.
• It is possible that a very popular website (like youtube) may generate large amount of data. In that case we need to further shard the data (e.g. by timestamp, or sub-sites), but it will complicate the query.

Log processing service

• Data enrichment should use caching to optimize
• For example, it may need to join with User table, in which case user info should be cached in-memory like in redis
• It may need to look up session id, in which case previous actions in the same session should be cached.

Aggregation service

• For real-time analytics, consider introducing RT processing engine like Flink or Spark Streaming
• For non-RT cases, we can leverage periodic Spark jobs.
• How to deal with late data: watermark, and manually triggered backfill

Trade offs/Tech choices

• For "raw" data, we choose a no-SQL db like Cassandra or DynamoDB because the need to for high write throughput, and that the data is largely unstructured
• For aggregated/curated data, we choose a relational database. Alternative we can choose an OLAP DB like Druid or ClickHouse if we want to support flexible queries.

Failure scenarios/bottlenecks

IngestService, LogProcessingService, ReportService are stateless and can be horizontally scaled.

The LogQueue ingestion rate is ~1M * 1k * 20 / 86400 = 20k/sec and data throughput is ~200k/sec. This is not enormous but should should shard the message queue by customer as well.

There might be a spurt of high traffic. In that case the data pipeline should gracefully degrade

• The client can sample the data instead of sending everything, or maybe aggregate data on the client side. Some metrics will be missing but it is better than breaking the service completely
• We store the raw log without processing, and process them later.
• RT processing slows down

We should have monitoring and alerting in place.

All database should be high availability mode, have replicas.

Future improvements

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