通用数据架构思考（二）：异步任务系统设计

这是「通用数据架构思考」系列的第二篇。上一篇聊到，多模态数据如何从 Edge 稳稳地上传到 Cloud：对象存储放大文件，控制面保存元数据与上传状态，事件把后续动作从上传接口中拆出来。

但 READY 并不是终点。对一个视频、点云包或传感器片段来说，上传成功往往只是它第一次真正进入系统。接下来可能还要解码、抽帧、跑 GPU 推理、做质量检查、生成缩略图和索引，最后才成为一个可检索、可训练、可复用的数据集样本。

这些事情如果都塞进上传请求里，用户大概会先看到一个转圈十分钟的页面，然后开始怀疑系统是不是已经沉默地挂掉了。更现实的是：GPU 有限、任务会失败、依赖会变化、数据量会突然暴涨。于是这一篇顺着「数据上传」往下走：先说明异步任务调度器在做什么，再把数据处理作为一个具体例子展开。

异步任务调度器，究竟在调度什么？

它不是单纯的「消息队列 + 几个 worker」。消息队列解决的是把工作交出去；异步任务调度器负责把一次业务请求从「被接住」可靠地推进到「完成、失败或取消」。在这中间，它至少要持续回答四个问题：

1. 这项工作是否已被可靠记录，且能被用户查询？
2. 现在是否满足开始条件，例如依赖是否完成、配额是否允许？
3. 应该交给哪一类执行者，例如 CPU、GPU 或外部服务？
4. 执行中断、重复、超时或失败后，下一步是什么？

可以把它看成位于业务 API 与执行资源之间的一层控制面：

API / Event ---> Scheduler <------> State Store
                   |                  (Job、Task、依赖、重试、结果)
                   v
             Dispatcher / Queue -----> CPU Worker / GPU Worker / External Service
                   ^                             |
                   +--------- 状态与产物回写 ------+

这张图里，队列是传递执行通知的通道，worker 是干活的人；调度器则是那个持续盯着状态、依赖和资源约束的人。下面从上传后的数据处理出发，逐步把这层控制面拆开。

上传完成，不等于处理完成

先看一个非常常见的处理链路：

20GB 视频 / 点云包上传完成
  -> 解码与格式校验
  -> 抽帧、切片、生成预览
  -> GPU 推理或特征提取
  -> 自动标注 / 人工标注准备
  -> 质量检查
  -> 写入 Dataset Manifest 与索引

其中任何一步都可能耗时数分钟到数小时，也可能需要不同资源：CPU worker、GPU worker、专门的标注服务，甚至人工回调。它们显然不该占着 HTTP 连接等待结果。

因此，接口的职责应该很克制：接受一个「处理意图」，持久化为 Job，然后快速返回。

POST /processing-jobs

{
  "object_id": "obj_01H...",
  "pipeline": "dataset-ingestion-v3",
  "idempotency_key": "client-generated-uuid"
}

HTTP/1.1 202 Accepted
{
  "job_id": "job_01J...",
  "status": "PENDING"
}

这里的 202 Accepted 表达的是「我已接受处理」，而不是「事情已经做完」。只有服务在返回前已提交 Job 与去重记录的事务时，才能把它理解为「可靠接住」。客户端超时后重发请求也很常见，因此应以 idempotency_key，或 (tenant_id, object_id, pipeline_version, request_key) 之类的唯一约束返回原来的 Job，而不是创建第二份工作。调用方可以轮询 Job，也可以订阅状态事件或 Webhook；无论哪种方式，请求生命周期和任务生命周期已经分开了。

从调度器的视角，这次数据处理首先是一张有依赖的 DAG（有向无环图）：

             decode
            /      \
       extract-A  extract-B
           |          |
      inference-A inference-B
            \      /
          quality-check
                |
          dataset-manifest

整张图是一次完整的业务工作；图上的每一个节点，才是可以被领取和执行的动作。这样就自然引出 Job 与 Task 两层概念。

从一张 DAG，拆出 Job 和 Task

有了上述视角，再看 POST /processing-jobs 就自然一些：整张图是一项 Job，节点是可执行的 Task。调度器先接住一次完整的业务工作，再决定怎样拆分与派发。

• Job：用户或上游业务关心的一次完整工作，例如「把这批采集数据变成可用数据集」。它有业务身份、状态、创建者、审计信息和最终结果。
• Task：调度器实际派发的最小执行单元，例如「解析第 37 个分片」「为这段视频抽帧」「对这一组图像运行一次模型」。

Job 是给业务看的，Task 是给系统跑的。一个 Job 可以展开为许多 Task；Task 重试、扩容、迁移到别的 worker 时，Job 的语义仍然保持稳定。

Task 切得太粗，失败一次就得重跑很大一段；切得太细，调度、存储与消息本身会吃掉大量成本。没有放之四海皆准的粒度，但可以用两个问题校准：

1. 这一步失败后，重做它的成本能接受吗？
2. 这一步的运行时间和资源需求，是否足够相似，能交给同一类 worker？

例如一个 20GB 视频的「抽帧」可以按时间段切成 Task；而写 Dataset Manifest 通常应是一个较小、可幂等的汇总 Task。不要为了形式上的并行，把每一帧都变成一个独立消息——一分钟 30 帧的视频很快就会把一个看似简单的 Job 膨胀成数万条 Task 记录与消息。

这个划分同样适用于非数据业务。比如一次退款申请是 Job；风控校验、库存回补、支付退款和通知用户是 Task。支付回调或人工复核需要等待外部世界时，它又会自然走向 workflow；实时欺诈信号则可以由流处理系统独立计算。

业务状态、执行待办与数据本体

在数据平台里，一个很容易混淆的地方是：数据库和消息队列都像是在「排队」，但它们回答的问题不同。

Task queue 和 Kafka 都能驱动 worker，但不该被当作可以随意互换的东西：前者更直接表达「一项工作由一个执行者领取」，后者更擅长可保留、可回放的分区事件。无论选哪一种，把 Job/Task 的权威状态只放在消息系统里，等到需要回答「这个任务为什么被取消」「谁发起了它」「它第几次重试」时，往往就会想念一份可查询、可事务更新的业务状态。

更稳妥的方式是：先在 PostgreSQL 的事务中写入 Task 状态和一条 outbox 事件，再由 relay 把事件发布到 Kafka。这样数据库中的业务承诺，与「准备通知 worker」这件事不会各写各的、互相失联。

API / Scheduler
  -> PostgreSQL transaction
       - task: PENDING -> DISPATCH_PENDING
       - outbox: TaskReady(task_id)
  -> Outbox relay
  -> Kafka topic / task queue
       - task: DISPATCH_PENDING -> QUEUED
  -> Worker

DISPATCH_PENDING 表示「已经具备执行条件，但 relay 尚未确认消息已发布」；QUEUED 才表示已进入 broker 的待领取状态。实现上也可以把 QUEUED 定义为「待派发」而不再区分两个状态，但一定要把语义写清楚，避免排障时出现状态说已入队、broker 中却没有消息的误解。

这不是「全链路 exactly-once」魔法。relay 和 consumer 依然可能重复投递、重复消费；真正的落点是：把执行设计成 at-least-once，把状态更新和产物写入设计成幂等。 worker 应先幂等地落库结果和 Task 状态，再确认消息或提交 offset；如果在此前崩溃，重新投递是预期行为。

例如可以用 input_version + pipeline_version + normalized_parameters + output_kind 构成确定性产物身份；业务允许「同一输入主动重跑」时，再加 run_id 或请求幂等键区分意图。同一 Task 被多跑一次时，后来的 worker 不应再生成第二份成功结果，而是发现结果已经存在后安全退出。涉及付款、发货或第三方 API 时，数据库唯一约束只能保护自己的库：还要传稳定的 idempotency_key，或保存请求标识后回查；无法安全重试的动作则需要显式补偿。

Queue 的价值，不只是异步

Queue 最有价值的地方，是把生产速度与消费能力隔开。

设备可能在某个时段集中上传；GPU worker 也可能因为模型升级、节点维护或成本控制而暂时缩容。如果 API 必须立即找到空闲 GPU 才能返回，任何短暂波动都会直接变成用户请求失败。队列让系统先承认任务，再在资源允许时执行它。

不过队列不是无限容量的黑洞。积压意味着等待时间，等待意味着对象保留成本、用户体验和可能过期的业务价值。因此调度器必须有明确的背压策略：

• 按 tenant、数据源或任务类型设置配额与并发上限；
• 先做轻量校验，拒绝明显不合法或超预算的 Job；
• 用优先级区分在线预览、紧急回放与离线批处理；
• 队列长度、最老消息年龄和 consumer lag 超阈值时，触发限流、扩容或降级。

这里一个很朴素的原则很有用：系统在满载时，也要能说清楚谁在等、为什么等、预计要等多久。

失败才是调度器的日常工作

真正的异步系统，不是假设 Task 会成功，而是假设它随时可能失败：容器被驱逐、worker 进程卡死、模型服务超时、对象暂时不可读、下游限流，或者任务代码本身有 bug。

常见的状态机可以从简单版本开始：

PENDING -> QUEUED -> RUNNING -> SUCCEEDED
                      |    |
                      |    -> RETRY_WAIT -> QUEUED
                      |
                      -> FAILED / CANCELLED / DEAD_LETTER

几个工程细节决定它是否真的可靠：

1. 租约（lease）、心跳与围栏。 Worker 领取 Task 时拿到有过期时间的 lease；持续心跳续租。worker 消失后，lease 自然到期，Task 才能被别人接手。每次领取还应原子递增 lease_version 或 attempt；状态回写使用类似 WHERE task_id = ? AND lease_version = ? AND status = 'RUNNING' 的条件更新。这样旧 worker 即使在超时后恢复，也不能覆盖新 worker 的结果。
2. 带抖动的指数退避。 短暂网络问题不值得立刻把同一个服务打得更崩。重试间隔应递增并加入随机抖动；不可重试的参数错误则应直接失败。
3. Dead Letter Queue（DLQ）。 到达最大次数的任务不该无限重试。DLQ 是隔离与诊断入口，不是失败事实的权威来源；失败上下文、输入版本和错误分类仍要回写 Task Store。修复代码、输入或依赖后再创建明确的 replay attempt，而不是把旧消息原样塞回主队列。
4. 幂等而非侥幸。 消费到消息、开始执行、写结果、确认消息，这几个动作无法被一个普通分布式事务完全包起来。重复是常态，幂等键、条件更新和不可变产物版本才是保险。

取消也不是把状态改成 CANCELLED 就结束：排队中的 Task 可以停止派发；运行中的 Task 通常需要 worker 协作式地检查 cancellation flag 并清理临时资源。已经完成的不可逆外部操作无法被撤回，只能走明确的补偿步骤。

进度也值得单独设计。不要让 worker 以每处理一帧就写一次 PostgreSQL 的方式「直播」进度；可以按时间或比例节流上报，再由 Job 聚合器计算整体进度。对用户来说，42% 当然不错，但 正在对 8/20 个分片进行特征提取，预计还需 6 分钟 通常更让人安心。

让 DAG 持续推进

一旦「质检必须等推理完成」「Manifest 必须等所有分片完成」出现，单纯的 FIFO 队列就不够描述系统了。图上的边代表依赖，调度器只把依赖都满足的 Task 置为 READY 并投递。父节点失败时，可以选择让下游跳过、走替代分支，或让整个 Job 失败；这些都应该是 pipeline 的显式策略，而不是 worker 临场猜测。

Apache Airflow 和 Argo Workflows 都把 DAG 当作描述任务依赖关系的核心模型。对数据处理而言，DAG 的好处很直观：它把并行空间暴露出来，也让失败影响范围清清楚楚。

但也别把 DAG 神化，或者把它和 workflow 当成互斥选项。DAG 描述任务间的依赖图；workflow 描述流程如何在时间、事件、失败与补偿中推进。 一个 workflow 完全可以包含一段 DAG，DAG 编排器也可能提供有限的重试、条件和定时能力。只是当流程出现「等待人工审核两天」「等待第三方回调」「按业务规则补偿」「等待定时器」这类长时间、带状态的交互时，durable workflow 往往是更贴切的主抽象。

它和 Spark Batch 是什么关系？

这个联想是对的，但要稍微收紧一下：当一个 Job 的输入是有限的，并且它的 Task 要等上游依赖完成才启动时，它当然具有 batch job 的形态。前面的「解码 -> 抽帧 -> 推理 -> 质检 -> 建集」就是一批有边界的数据处理；整个 Job 有明确开始和结束，依赖满足后释放下游并行度，这和 Spark 的执行图很像。

Spark 也会把一次并行计算拆成多个 Task，并按依赖切分为 Stage；同一 Stage 中的 Task 往往并行处理不同分区，shuffle 等边界会让下游 Stage 等上游结果可用后再运行。Spark 的官方术语中，Job 是由 action 触发的一组并行计算。这正是「依赖满足后再执行」的计算侧版本。

差别在于，Spark 的中心问题是：怎样高效执行一个数据并行计算图。 它擅长把大量相似的数据分区送到 executor，处理 shuffle、资源分配和失败重算。通用异步调度器的中心问题则更宽：怎样把一个业务承诺可靠地推进到完成。 Task 可能不是计算，而是申请上传会话、调第三方 API、等待人工审核、发通知或启动一个 Spark application；它还要保存用户可见状态、处理取消和补偿，并跨越很长的等待时间。

因此，把它们看成同一条谱系上的两层会更准确：Spark 是批量数据计算的专用调度与执行引擎；通用调度器是面向业务流程的编排层。 实践中常常二者并用：通用调度器把「运行某批数据的特征提取」当作一个 Task，提交 Spark 作业，等待其成功或失败，再继续质量检查和 Dataset Manifest。Spark 内部再负责把这项计算拆成 Stage 和 Task，并把资源分给 executor；它也有自己的 job scheduling 与资源分配机制。

Queue Worker、DAG、Workflow 与流处理，究竟怎么选？

这四种主要抽象 / 运行模式都能「异步做事」，因此经常被混着讲；但它们对输入、依赖和状态的假设并不一样。一个速记是：独立动作 -> Queue；有限依赖 -> 批处理 DAG；等待外部世界 -> Workflow；无界持续数据 -> Stream。 有界/无界只是第一层判断，之后还要看结果是否持续更新、允许的端到端延迟、是否依赖 event time，以及状态能否按窗口自然收敛。

框架的能力会跨越这些边界：例如 Argo 是 workflow 引擎且支持 DAG，Spark Structured Streaming 与 Beam 更接近流计算运行时 / API。这里按最常见的使用方式归类，不把它们看成可以一键互换的产品清单。

1. Queue + Worker：先把独立工作从请求线程拿走

这是最常见、也最值得先从这里开始的模型：生产者把 Task 放进队列，worker 竞争领取并执行。worker 不需要知道完整业务流程，只要知道「这条消息对应什么工作、如何安全完成它」。

例如对象进入 READY 后，投递 generate_thumbnail(object_id)；worker 下载视频、生成缩略图、将结果写到对象存储，然后更新 Task 状态。邮件、图片压缩、Webhook 投递、导出报表也都很适合这一层。

Python 的 Celery 就是这个模式的典型实现：client 通过 broker 把 Task 消息送给 worker，worker 持续从队列领取任务；broker 可用 RabbitMQ、Redis 等。Celery 官方的定义就是 task queue 用于跨线程或机器分发工作。

Celery 也不只会「扔一个独立任务」。它的 Canvas 提供 chain（串行）、group（并行）和 chord（一组任务完成后执行 callback）等组合原语；chord 已经很像一个小型 DAG 的 join 节点。Celery Canvas 文档有这些语义的具体定义。因此更准确的说法是：Celery 以 Queue + Worker 为核心，可以拼出不少 DAG 式工作；但复杂业务流程的持久化、可视化、版本演进和补偿，仍需要由应用自己补齐。

适合 Celery 的信号通常是：任务以秒到分钟计、依赖不复杂、失败后重试即可、业务状态已经由自己的数据库保存。要特别保留前面说的幂等、重试和 DLQ；换成 Celery 并不会让这些问题自动消失。

2. DAG Scheduler：让依赖和并行度成为一等公民

当一个 Job 不是「做一个动作」，而是「做完一批相互依赖的动作」时，DAG 会更清楚。调度器知道每个节点的上游；上游全成功才释放下游，彼此无依赖的节点则可以并行。

拿多模态数据举例：decode 成功后可以同时启动按分片划分的 extract-frame-*；所有分片完成后才开始 inference；最终由 dataset-manifest 汇总产物。这不是让一个 worker 自己阻塞等待，而是调度器根据依赖图不断把新的 ready task 放到执行队列。

Airflow、Argo Workflows 都属于这类思路。它们尤其适合每日 ETL、离线特征计算、模型训练或有明确输入边界的数据集生产：你希望看到整张图、知道哪个节点失败、只重跑受影响的分支。

代价是 DAG 必须被维护成真正的契约：Task 输入/输出、失败传播、重跑是否安全、动态分片如何展开，都要定义清楚。它并不天然适合「用户三天后点击确认」这样的长等待；把一个 DAG 节点长期占着 worker 等人，通常是一个危险信号。

3. Durable Workflow：把等待外部世界也变成可恢复状态

Durable Workflow 处理的不是计算依赖本身，而是业务过程中的时间与不确定性：等人工、等支付回调、等设备再次上线、30 分钟后重试、失败后做补偿。它会把流程的推进历史持久化，因此 worker 或服务重启后，不需要靠一段常驻进程「记住自己等到了哪一步」。

例如一个数据集发布流程可以是：发起发布 -> 等质量负责人审批 -> 提交合规扫描 -> 等扫描服务回调 -> 发布索引；审批被拒绝时撤销临时访问权限。另一个典型例子是订单履约：扣款成功后等待仓库确认；如果超时，则按明确策略退款或人工介入。

Temporal 的 Workflow 模型就是这一类思路：服务端保存 execution history，worker 故障后会通过回放历史重新执行确定性的 Workflow 代码，而不是取回某个 worker 的内存快照。网络调用等非确定性工作应放在 Activity 中；补偿也必须由业务显式建模。选择它的理由不是「任务有依赖」，而是流程需要跨很长时间、接收外部信号、可审计地重试与补偿。

4. Stream Processing：数据永远还在来

流处理的根本区别在于：输入不是某个 Job 的有限文件列表，而是一条可能永不结束的事件流。系统不是等「全部处理完成」，而是持续维护一个随新事件变化的结果。

例如每秒到来的设备 telemetry，可以按设备和五分钟窗口计算平均温度、异常比例与在线状态；实时风控可以把支付事件与最近行为拼起来，持续更新风险分数。事件会迟到、会乱序，因此需要 event time、水位线和窗口；执行器重启时，还要用 checkpoint 恢复已维护的状态。Apache Flink 的 state 与 checkpoint围绕的就是这种持续计算。checkpoint 让状态恢复成为可能；端到端 exactly-once 仍取决于 source、sink 与 checkpoint 协议能否共同支持，不能把它理解为所有外部写入都天然精确一次。

这里也可以有 DAG，但它描述的是一条持续运行的数据流图，而不是「跑完就结束」的 Job 图。把设备 heartbeat 交给离线 DAG，每天才算一次在线状态，通常太慢；反过来，把每次离线建集都塞进持续流作业，也会让重跑、版本化和成本控制变复杂。

一个现实的组合

真实系统往往不是四选一，而是各自放在最合适的位置：

对象 READY
  -> Queue + Worker：轻量校验、创建 Job、投递处理请求
  -> DAG / Spark Batch：转码、分片推理、质量检查、建集
  -> Durable Workflow：等待人工审批、外部扫描回调、发布或补偿

设备持续上报 telemetry
  -> Stream Processing：窗口聚合、在线状态、实时告警

所以不要因为系统里已经有 Kafka，就断言所有东西都该变成流处理；也不要看到 Task 有依赖，就一定引入一套重量级工作流引擎。先识别任务的时间边界与业务语义，再选择模型；底层消息队列可以复用，语义不必强行统一。

调度其实有好几层

还有一个容易被忽略的事实：所谓「调度器」常常不是一个组件，而是几层不同的决定叠在一起。

业务调度层：哪些 Job / Task 可以开始？优先级是什么？（现在该不该跑？）
        |
执行调度层：哪个 worker group 来领？CPU 还是 GPU？并发多少？（谁来跑？）
        |
基础设施层：Kubernetes 把 Pod 放到哪台机器？能否拿到 GPU、磁盘和网络？（跑在哪？）

Kafka consumer group 负责消息分区的消费协调，Kubernetes scheduler 负责 Pod 放置，workflow/DAG scheduler 负责业务依赖。它们边界不同，谁都不能完全替代谁。把业务优先级、租户配额和「这个 Task 是否可运行」塞给 Kubernetes，通常会让系统很难解释；反过来让业务调度器假装自己能直接管理每块 GPU，也会制造另一套资源管理系统。

回到 Edge-to-Cloud：一条完整的数据处理路径

把前一篇的上传路径接上，整条链路可以这样理解：

Edge 采集并上传对象
  -> 控制面验证 checksum / session，object 进入 READY
  -> 创建 ingestion Job 与初始 Task（同一事务写 outbox）
  -> 事件进入执行队列
  -> CPU / GPU worker 按依赖消费并生成版本化产物
  -> 质量检查与元数据索引
  -> Dataset Manifest 记录 input、transform、processor version
  -> Job SUCCEEDED，数据可以被检索、回放或用于训练

关键不是每一步都必须用同一种技术，而是每一层都有清晰的权威来源：对象存储保存原始输入与产物，数据库保存业务状态与谱系，消息系统承载可重放的执行通知，worker 只负责把一个明确的 Task 推进到下一个确定状态。

从这个角度看，数据处理调度器和通用异步任务系统是同一道题：前者只是任务比较偏向解码、推理、质检和建集；后者可能换成发邮件、生成报表、对账、调用第三方服务，底层仍然面对可靠接单、排队、依赖、失败、重试、幂等和可观测性。

一个够用的落地顺序

不必一开始就实现 DAG、优先级抢占、跨地域容灾和复杂补偿。一个更实际的演进路径是：

1. 先有 Job + Task 表、明确状态机、可查询的接口；
2. 用 outbox 接上队列和 worker，完成 可靠投递、at-least-once 与幂等；
3. 加 lease、心跳、重试、DLQ、指标和失败对账，做到 可恢复；
4. 任务确实出现依赖时，再引入 DAG 与 ready-task 调度，获得 可编排 的批处理；
5. 出现长时间外部等待或补偿时，评估 durable workflow；实时无界数据则交给 stream processing，处理 长流程 / 实时流。

这套顺序的好处是，前几步已经能解决绝大多数「上传之后怎么办」的问题；后面的复杂度，是被真实的业务形态推着长出来的，而不是为了架构图好看提前买单。

小结

异步调度系统的核心不只是「开一个队列、起几个 worker」。它是在不可靠的执行环境中，对业务做出可靠承诺：任务不会无故丢失；重复执行不会污染结果；失败能恢复也能解释；资源紧张时系统知道该让谁先跑；数据处理完成后，结果和来路都查得到。

上一篇解决的是数据怎样进入云端；这一篇解决的是数据进入之后怎样被可靠地加工。无论是企业报表还是大规模多模态数据集生产，面对的都会是同一组朴素但绕不开的问题。

References

• Apache Airflow — DAGs
• Argo Workflows — DAG
• Temporal — Workflows
• Apache Flink — Working with State
• Apache Spark — Cluster Mode Overview
• Apache Spark — Job Scheduling
• Celery — Introduction
• Celery — Canvas: Designing Work-flows
• Martin Kleppmann, Designing Data-Intensive Applications