系统设计：如何设计一个分布式作业调度器？- Rakshesh

工作调度是一个众所周知的系统设计面试问题。下面是一些可能需要设计工作调度器的领域。

设计一个付款处理的系统。(即每月/每周/每天的支付等)
设计一个代码部署系统。(即代码流水线)

这个职位的目的是设计一个简单但可扩展的作业调度系统。

问题说明

设计一个作业调度器，在预定的时间间隔内运行作业

功能要求

用户可以安排或查看工作。
用户可以列出所有已提交的作业和当前状态。
工作可以运行一次或重复运行。工作应在定义的计划时间后的X阈值时间内执行。(让x=15分钟)
单个作业的执行时间不超过X分钟（让X=5分钟）。
工作也可以有优先权。优先级较高的作业应该比优先级较低的作业先执行。
工作的输出应该存储在文件系统中

非功能要求

高度可用 - 系统应始终可供用户添加/查看作业。
高度可扩展性--系统应可扩展至数百万个作业
可靠性--系统必须至少执行一次作业，而且同一作业不能由不同的进程同时运行。
持久性--在发生任何故障时，系统不应丢失作业信息。
延迟性 - 系统应该在作业被接受后立即确认用户。用户不需要等到作业完成。

系统接口
有三个API可以暴露给用户

1. submitJob(api_key, user_id, job_schedule_time, job_type, priority, result_location)
这里，job_type = ONCE或RECURRING，result_location可以是s3位置。
API在接受作业后可以返回http响应代码202

2. viewJob(api_key, user_id, job_id)
响应包括状态为NOT_STARTED、STARTED或COMPLETED

3. listJobs(api_key, user_id, pagination_token)
用户可以查询所有提交的工作，并返回一个分页的响应

用户请求流程

用户通过连接负载均衡器（或API网关）来提交/获取工作。
请求将持续存在于数据库中，并将确认函发回给用户
工作执行者服务将不断从数据库中轮询到期的工作，并在队列中保持输入。
工作执行者服务将执行实际的工作业务逻辑，并将最终结果更新到文件系统中，并将状态更新为完成。

数据库设计
由于我们没有严格要求交易支持或任何其他ACID属性，并考虑到峰值QPS（2*1000=2000 QPS），我们可以使用SQL或NoSql数据库。考虑到NoSql在规模、维护和成本方面的明显优势，我将使用DynamoDb的NoSql解决方案。

用户查询模式。

给定userId，添加工作
给定用户身份，检索所有工作编号

数据库表结构：

Table: JOB

+------------------------------+--------+
|          Attribute           |  Type  |
+------------------------------+--------+
| user_id (partition key)      | uuid   |
| job_id (sort key)            | uuid   |
| actual_job_execution_time    | date   |
| job_status                   | string |
| job_type                     | string |
| job_interval                 | int    |
| result_location              | string |
| current_retries              | int    |
| max_retries                  | int    |
| scheduled_job_execution_time | date   |
| execution_status             | string |
+------------------------------+--------+

job_status。这是用户将看到的工作状态。它可能有：NOT_STARTED, STARTED, COMPLETED

execution_status。这是我们的服务将保持的实际执行状态。它可能有。not_started, claimed, processing, success, retriable_failure, fatal_failure

除了用户之外，我们的工作调度服务将轮询数据库以获得到期的任务。我们有不同的方法来实现这一点

1. 基于X分钟大小的桶窗口进行分区
我们可以创建索引，命名为 scheduledJob，以检索最后X分钟到期的工作。

Index: ScheduledJob
+----------------------------------------------+------+
|                  Attribute                   | Type |
+----------------------------------------------+------+
| scheduled_job_execution_time (partition key) | time |
| job_id (sort key)                            | uuid |
+----------------------------------------------+------+
Query (SQL equivalent):
SELECT * FROM ScheduledJob WHERE scheduled_job_execution_time == now() - X

2. 基于X分钟大小的桶窗口加分片ID的分区
有可能在一个特定的时间窗口，已经收到了许多作业（比方说100K）。在这种情况下，上述查询性能将非常缓慢。我们可以根据随机的（比方说在1到Y之间）shard_id来进一步划分数据库。

Index: ScheduledJob

+----------------------------------------------+------+
|                  Attribute                   | Type |
+----------------------------------------------+------+
| scheduled_job_execution_time (partition key) | uuid |
| shard_id (partition key)                     | int  |
| job_id (sort key)                            | uuid |
+----------------------------------------------+------+
Query (SQL equivalent):
SELECT * FROM ScheduledJob WHERE scheduled_job_execution_time == now() - X and shard_id == Y

工作调度器是如何工作的？
作业调度流程：
每隔X分钟，主节点会创建一个权威的UNIX时间戳，并给每个工作节点分配一个shard_id和scheduled_job_execution_time。
工作者节点将执行以下查询，并将作业推送到Kafka队列中进行执行。

Worker 1:
SELECT * FROM ScheduledJob WHERE scheduled_job_execution_time == now() - X and shard_id = 1
Worker 2:
SELECT * FROM ScheduledJob WHERE scheduled_job_execution_time == now() - X and shard_id = 2

容错：
Master监控工作者的健康状况，知道哪个工作者死亡，以及如何将查询重新分配给一个新的工作者。
如果主节点死亡，我们可以分配其他工作节点作为主节点。
此外，我们还可以引入本地数据库来跟踪工人是否成功查询了数据库并将条目放入队列中。

工作执行器是如何工作的？
作业执行器服务有多个消费者，从队列中提取数据。消费者机器也有主进程和工作进程。主进程和工作进程都是基于 "拉 "的模式进行操作。主进程将从队列中轮询作业，工人进程将通过执行以下代码从主进程中持续轮询作业

while True:
  w = get_next_work()
  do_work(w)

作业执行流程和容错

当一个作业从队列中被取走时，消费者的主站会更新JOB db属性execution_status=CLAIMED。
当worker进程拿起工作时，它更新execution_status=PROCESSING，并不断向本地数据库发送健康检查。
工作完成后，worker进程将把结果推送到s3，更新JOB数据库的execution_status=COMPLETED或FATAL_FAILED，以及本地数据库的状态。
worker进程和主站都会在本地数据库中更新健康检查。

健康检查器服务
健康检查器服务定期运行（例如每隔 x 秒），并扫描最后一次从工人进程收到的健康检查低于定义阈值的数据库。在这种情况下，它认为该作业未能处理，并将条目推回队列。

系统设计：如何设计一个分布式作业调度器 ？- Rakshesh

系统设计：如何设计一个分布式作业调度器？- Rakshesh