System requirements

Functional:

1. Process both real-time and batch data.
2. Data come from structured source (e.g. database), less structured source (e.g. raw vendor files), and unstructured source (e.g. website).
3. Historical data is kept.

Non-Functional:

1. Depending on the use cases, we need to choose between high availability vs high consistency. Assume we choose high consistency for services like financial services.
2. Need backups / redundancy if the main node fails.
3. Data integrity: data shall not be altered by any unauthorized access.
4. Expiration: data older than 1 year is removed from storage.
5. monitoring and alerting: if applicable, monitoring data abnormaly and traffic abnormaly.

Capacity estimation

1. Assume daily incoming data is 1 TB. Write QPS = 1 TB / 86400 = 10 MB / sec. Peak Write QPS = 10 MB * 3 = 30 MB / sec. This is below modern network bandwidth, and also below modern SSD write speed limit.
2. Assume we keep 1 year history. Total storage = 1 TB * 365 = 365 TB. This is possible to store in a SSD or HDD.

Although the limits may be handled by a single server, we need redundancy in case the main server fails.

Also, many data require processing and may have intermediate outputs, which consumes CPU bandwidth and also reduces disk available bandwidth. We may need multiple servers to handle different data.

API design

Authentication:

Auth authenticate(user, access);

• user: who is to use the data pipeline
• access: an object to indicate which data the user needs access and which ops (write or read) the user needs
• return an Auth object, which may contain an auth token

Write and Read accesses are only available after the user is authenticated.

Write:

int write(input, processing_rules, destination);

• input: data input
• processing_rules: rules to process data to a format to be stored
• destination: database or other storage where processed data is stored
• return code is int, to indicate write result status

Read:

int read(&output, source, data_requested);

• output: the formatted output data that is read
• source: where to read data from
• data_requested: what data is to read
• return code is int, to indicate read result status

Database design

Depending on the data characteristics, different databases are chosen.

SQL: structured relational data, e.g. user profile

Document DB: e.g. MongoDB, to store documents, such as news feed, etc.

Append-only DB: e.g. Cassandra, to store structured analytical data

Time series DB: e.g. kdb, to store financial data

High-level design

See diagram

Request flows

Write:

1. Real-time: data goes to load balancer. Load balancer decides which server to send data to process based on data characteristics. RT processing server processes the data and write to DB.
2. Batch: Batch load balancer gets request from batch scheduler or directly from batch data input source. Load balancer sends data to batch processing server. Batch processing server processes the data and write to DB.

Read:

request goes to load balancer, then route to read server. Server reads data from cache, or from DB on cache miss.

Archive:

Scheduler kicks off archiving process and move older data to archive storage.

Detailed component design

1. Archive: this may be multi-tiered storage. Older data and less frequently used data may be moved to lower tier / cold storage. For example, AWS S3 has regular tier for regular access and multiple glacier tiers for less frequently used data. It's ok for the archiving job to fail and retry.
2. Write: each server may handle different data sources. Each server has multiple CPU cores, and thus different datasets shall be processed by different CPU cores to maximize computing power. Each server may need a redundancy server to stand by in case of failure. The final processing throughput is bound by disk I/O (if writing to disk) and/or database write speed (if writing to DB). We might need a queue, to temporarily hold processed data to be written, between the processing server and DB if the DB write speed is significantly limited compared to server processing speed.
3. Monitoring and alerting: this may be put on processing server to run by different CPU core. Rules shall be configured to indicate what conditions are abnormal and what actions to take.
4. Authentication: this may be done by registering user profile/credentials to a separate user DB, with different level of ACL. There will be a separate system that handles reviewing and granting ACLs to new users, or reviewing and updating ACLs to existing users.

Trade offs/Tech choices

We separate read and write to different servers because the data ingestion throughput is bound by disk I/O and/or DB write throughput. If read and write are put on the same server, we still have enough CPU power to process the requests and data, but read speed might be limited if disk is involved.

We may also need DB replicas in case of DB failure. DB shall have a write-ahead log to avoid possible data duplication or corruption when the DB fails; or some other idempotent mechanism shall be implemented on the processing server side to ensure such data integrity.

For the queue between write processing servers and DB, it may be partitioned based on which DB the data is going to, or partitioned based on server. The former may be cleaner because one single queue contains all same kind of data, and since one kind of data is processed by one server, the order is guaranteed same as data initial arrival, which may be critical in some use cases, such as financial market data.

Failure scenarios/bottlenecks

If a specific kind of data is large, the server (and the queue) which is handling this dataset may become computational bottleneck. We'll need to distribute the traffic to multiple servers. In this case, if the order of data matters, we need to aggregate them per their original order. We may rely on the load balancer, which is initially handling the requests, to put a timestamp or sequence number on each of the incoming request.

We choose consistency over availability, in which case we can retry if the read and/or archiving job fails on the first attempt.

Future improvements

If the data size is much larger than 1 TB daily, then a single dataset may not be handled by a single server. We'll distribute such traffic to multiple servers.