System requirements

Functional:

From data perspective:

1. data ingestion: should be able to support data ingestion through different types of platform: streaming, batch process, ODBC direct reads
2. data validation: should be able to support verification and validation for the data in terms of format and value, and eliminate any invalid data ingested
3. data normalization (formatting/cleaning/deduplication): should be able to take different formats of data and extract the core information after necessary pre-processing
4. data storage: should be able to store data in different stages: staging, pre-processed, business-ready in corresponding storage toolings, like S3, DB, etc
5. data analysis: should be able to provide reporting, narrative, visualization, even AI-powered analysis based on user's requirements
6. data feed generation and propagation: if possible, mature and customized data feed should be generated from the system and pass down to any consumers for further processing or purpose
7. system monitoring and logging: the system itself should have a control plane feature which monitors the system to ensure availability & performance, and warn stakeholders in case of any threshold breach

From user perspective:

1. user registration
2. user should be able to visualize and create DAGs to indicate specific flow for data from ingestion all the way to storage or further process
3. once flow is created for each task, it will be become scheduled jobs that one-time or periodically handles data workflow, from ingestion - process - storage - more.
4. system should support workflow creation, workflow trigger, job monitor, etc

Non-Functional:

1. availability: the system should be highly available with a uptime of > 99.99%
2. scalability: the system should be scaled to be able to handle high volume of concurrent requests
3. performance: each major component in this system (for data ingestion) should be able to handle requests with low latency or aligned with defined SLA
4. security: in cases where data include user-identifiable, confidential or highly sensitive data, the system should use encryption and necessary software to enhance and guarantee data security
5. reliability: each layer in the system should ensure data integrity and data analysis should generate accurate and reliable results
6. extensibility: in case of any new raw data needed to be ingested in new data form and using new data channel, the system should be able to easily extend to accept new dat schema and even new business requirement

Capacity estimation

Average storage estimation: Assume regardless of the data channel or format, there will be 10M pieces of data unit to be ingested in average in a day, then processed and stored by this system. Assume data of each type could differ in terms of size, but let's assume in average one piece of data unit is ~ 10KB. In a day we will expect a total of 100 GB size of data, that is 3TB size of data each month.

During daily peak hour, let's assume 10% of daily traffic will occur in this singular hour, that is 1M data transactions and 10 GB size of data to be ingested/processed in each hour, the TPS within that peak hour is 1M / 3600s = 277 ~ 300 TPS, and the system throughput within that peak hour is 10GB / 3600s = 10M * KB / 3600s ~= 3000 KB/s = 3MB/s.

API design

1. /usr/registration
   1. request body: user_name, user_email
   2. response body: user_id, created_at
2. /workflow/create
   1. request body: DAG nodes, directional edges, detailed configuration for each node: type, engine, description, script, etc
   2. response body: creation status, workflow id
3. /workflow/update
   1. request body: DAG nodes, directional edges, detailed configuration for each node: type, engine, description, script, etc
   2. response body: update_status, workflow id
4. /workflow/delete
   1. request body: workflow_id
   2. response body: delete_status
5. /job/create
   1. request body: workflow id, type: event trigger or cron based, schedule, recurring_flag
   2. response body: creation_status, job id, next_run_time, schedule_status
6. /job/schedule
   1. request body: job_id, trigger_time
   2. response body: schedule_status, trigger_time, job_status
7. /job/execute
   1. request body: job_id
   2. response body: execute_status

Database design

1. parameterized data (for example if it's a finance data system): invoice id, amount, customer name, product type, country, storefront, etc
2. user: user_id, user_email, user_name, created_at, hashed_pwd, etc
3. workflow: user_id, workflow_id, node_list, edge_list, created_at, updated_at, status
4. job: workflow_id, status, next_run_time
5. node: type, engine, description, script

High-level design

From data flow perspective: raw data (whether streaming or cron-based) -> bronze layer (ingested as-is, stored as raw data) -> silver layer (data normalization, including data validation/cleaning/deduplication/formatting, stored in staging storage like Amazon S3) -> gold layer (user customized logic: storage, ML logic, transform, analysis) -> reverse ETL, feed to downstream systems

From user perspective: web clients -> load balancer -> API gateway -> services (workflow service, scheduler service, monitor service, user management service) -> cache for highly frequently accessed data -> workflow config will be stored in Redis Sorted Set and then persisted to PostgresSQL -> Once a job is due, will be published to corresponding Kafka topics -> Kafka will be used for job dispatcher -> Kafka partitioned by userId -> once consumer consumes the job, will execute the job step by step -> for each step, pass the logic to corresponding engine runner and execute the next node

Request flows

From user perspective: web clients -> load balancer -> API gateway -> services (workflow service, scheduler service, monitor service, user management service) -> cache for highly frequently accessed data -> workflow config will be stored in Redis Sorted Set and then persisted to PostgresSQL -> Once a job is due, will be published to corresponding Kafka topics -> Kafka will be used for job dispatcher -> Kafka partitioned by userId -> once consumer consumes the job, will execute the job step by step -> for each step, pass the logic to corresponding engine runner and execute the next node

Detailed component design

Would love to dive into the workflow and job creation, scheduler, and execution. From the web UI, users will be able to drag and form a visualized DAG for the user-defined data workflow. A DAG will be consist of nodes and directed edges between nodes, each node represent a certain type of data operation, Spark, SQL, or just python script. Following the edges will be able to follow the workflow from begin to end. Once user finishes creating the DAG, the system will first evaluate and validate the DAG, once validated, a corresponding unit called job will be created, it's a schedul-able and executable unit of workflow. The DAG config will be stored in PostgresSQL, and the scheduled job will be added to Redis Sorted Set, next_trigger_date as the score. A poller server will periodically check the queue (Redis Sorted Set) for any due jobs, if found, poller server will publish these jobs to Kafka topics (if recurring, poller server will add a next job back into Redis Sorted Set). Underlying executor/worker as Kafka consumers will pick up these jobs and execute them one by one. For each node encountered in this job, executor will call upon corresponding engine runner to run this node, whether Spark, SQL or python. Each engine runner will be using Docker for encapsulation and env isolation.

Trade offs/Tech choices

1. I use SQL DB PostgresSQL for storing workflow config, because workflow could be complex with all the nodes and edges involved, SQL in general supports more complex queries compared to NoSQL. Also the DB for workflow config wouldn't be write-heavy instead it will be read-heavy and requires high-level of consistency in case of workflow changes. SQL is more suitable choice, but sacrificing some of the throughput and scalability ability.
2. By using PostgresSQL I'm leaning into forcing consistency over eventual consistency because workflow might be updated due to any business requirement and I want the update to be reflected as soon as possible, but with consistency there is a small price to pay for write latency.
3. For faster access for some highly frequently accessed data, I'm using Redis Cache, I will be enforcing write-through strategy to ensure consistency
4. For the job scheduling and execution I'm breaking the backend component into many micro-services, workflow service, scheduler service, dispatch service, executor service and engine running service, this will increase the cost and complexity in terms of managing and maintenance however the isolation and decoupling will assist in scalability.
5. I'm choosing Redis Sorted Set instead of directly writing into db, because in terms of finding next due jobs, I want to latency to be minimal, however due to in-memory volatile natural, sometimes some jobs might be lost or forgotten.
6. Which is why, I'm introducing a separate component which is an async rabbitMQ to detect any long-term due jobs in db and push them back to scheduler queue if need. This MQ will be also be used in other monitoring and retry.
7. I'm choosing Kafka because it naturally scales really well and high-throughput for fan-out the tasks works well with our possible giants volume of jobs
8. For each major component in this system, I will replicating most of them, it increase a bit of cost but it will provide great availability and fault tolerance, in terms of failure events, we can fall back to use replicates.
9. I'm using a combination of load balancer and API gateway because I want to ensure both: 1) load and traffic distribution for better concurrency and latency 2) API gateway to handle auth, rate limiting and navigation between varies micro-services.
10. I'm choosing Redis Cache over Memcache due to the complex data structure it supports and it scales better.

Failure scenarios/bottlenecks

1. Because the data being ingested by the system is completely user generated, the data validation sometimes might fail to capture all possible invalid scenario and produce proper error handling, in that case the system might fail while executing some of the nodes
2. In some of the nodes, if the engine is python, and user passes in a python script for execution, need to be extra careful for possible RCE security issue.

Future improvements

1. In terms of data-driven products, a lot of AI/ML products and tools can be combined to generate more user-customized and overall smarter analysis results and narratives. For example, could also incorporate a NL chatbot for asking data analysis questions and receiving NL response.