2026-06-05-云平台及遥操设计

我在机器人创新中心和鲸跃动力两个公司加起来差不多一年，前半部分做工程治理，云上基础架构，后半端就开始做云平台和遥操了。当然在鲸跃动力也干了不少不被老板认可的脏活累活，所以对一线工作有了不少的经验，写这个文档就是记录下大的工程建设

一、 总体设计

#

总体设计分为了三块，遥操平台，数据&ml平台，云管平台。为何这样子设计呢？因为功能和性能的矛盾过于明显

如果把「遥操、传视频、存 MCAP、训模型、发版 OTA」塞进一个单体服务，边界会迅速糊掉：MQTT 里塞训练参数、数据库里塞心跳、UI 直连 Kubernetes API——运维和迭代成本都会爆炸。所以最终设计为了三个平台

• 云管平台：回答「机器人是否在线、任务/OTA 怎么下发、失败了怎么回滚、如何调度」。
• 遥操作平台：回答「人如何低延迟看到画面、发出控制」——不走 Fleet 的任务状态机传每一帧。
• 数据&ML平台：回答「文件在哪、这次采集是什么语义、怎么 freeze 成 dataset、训练用了哪一版数据、如何训练」。

简单来说，理解为遥操作平台 & 数据&ML平台是基于云管平台的上层平台。云管平台只要保证了安全，可靠等底层问题，上层平台的功能封装为任务下发即可。

二、云管平台（K8S4R）

#

云管平台命名为 K8S4R（Kubernetes for Robots）。核心思想：用声明式范式管理机器人资源和任务——管理员声明「要什么」，系统持续调和到「实际是什么」，而不是为每台设备写一套脚本。

控制面规划两个版本，共享同一套运行时管道（MQTT → Kafka → Flink → Redis），只在控制面实现上分叉：

• 版本一：K8s CRD + Robot Manager + **Server（下发网关）**这个实际上不该选择kubernetes，所以最终干掉
• 版本二：自建 Fleet API + Controller Manager，运行时管道与版本一相同

2.1 组件架构

#

我将架构分为三个部分：

• 机器人端：mqtt客户端，支持本地持久化任务，状态上报等功能。
• 数据平面：系统的“神经系统”，由 MQTT Broker、Kafka、Flink、Redis、TSDB 组成。负责高可靠的机器人通信、实时数据聚合、状态快照存储。控制平面仅通过 Redis 读取实时状态，不直接处理高频消息流。
• 控制平面：运行在我们自己的服务上，由 CRD（Job, TaskGroup, Task, Robot）+ Manager（Operator）+ Server（指令下发组件）组成。负责业务逻辑、调度决策、状态管理，所有关键状态持久化在 pg数据库 中。

下面给出每个部分包含的架构：

• 机器人端：
  • MQTT客户端：MQTT 客户端，双向 TLS 认证，遗嘱消息 + 心跳（Keep Alive 30s）。订阅专属命令主题，QoS 1 保证送达。执行结果以业务回执（QoS 1）上报。高频遥测（位置、电量等）用 QoS 0，资源指标（CPU/内存/磁盘）定期上报。
  • SQLite关系型数据库，存储诸如任务状态等关键信息
• 数据平面：
  • 1. MQTT Broker 集群
    • 采用 EMQX 多节点集群，前端负载均衡。
    • 机器人通过 X.509 证书认证，ACL 控制每个客户端只能访问自己的主题。
    • 遗嘱消息与心跳配合，45 秒即可感知离线。
    • 规则引擎将全部遥测消息（包括心跳、任务回执、告警）实时桥接到 Kafka，过程中注入 W3C TraceContext 实现全链路追踪。
  • 2. Kafka 集群
    • 作为消息缓冲区，持久化保存所有原始消息（保留 7 天），解耦 Broker 与后端处理。
    • 规则引擎以 acks=all、幂等 Producer 写入，保证不丢消息；在 Kafka 不可用时支持磁盘缓冲和熔断。
  • 3. Flink 流处理
    • 从 Kafka 消费原始消息，进行：
      • 状态聚合：为每个机器人维护最新状态快照（位置、电量、在线状态、资源使用率），写入 Redis。
      • 事件提取：基于定时器和模式识别，生成“机器人离线”、“任务超时”、“CPU 过载”等业务事件，可写回 Kafka 的 robot.events 主题或直接触发告警。
      • 版本过滤与去重：利用消息序列号实现幂等更新，避免重复写入。
    • 同时将原始时序指标（如 CPU、内存历史）直接写入 TDengine 等 TSDB，供监控面板查询。
  • 4. Redis 集群
    • 存储所有机器人的实时瞬态数据：robot:status:{id} 中包含位置、电量、在线标志、资源使用率、当前执行任务 ID，以及空闲机器人 Sorted Set 等。
    • 数据带有 TTL（如 60 秒），由 Flink 持续刷新，仅作为查询缓存，不是权威存储。
  • 5. TSDB（TDengine/InfluxDB）纯理论，没做
    • 归档历史心跳、位置轨迹、资源使用率等时序数据，用于离线分析和容量规划。
• 控制平面：
  • CRD 定义
    • Job：顶层业务目标，含任务模板、完成条件。
    • TaskGroup：经调度拆分的批次，绑定到一组机器人（通过选择器）。
    • Task：最小执行单元，直接指定目标机器人、操作内容、超时、重试策略。Status 包含 phase（Pending/Scheduled/Dispatched/Running/Succeeded/Failed）、assignedRobot、messageAcked（表示指令是否已送达）等。
    • Robot：机器人期望状态（如 desiredVersion）和当前状态（online、currentTask、conditions 如 Healthy/MemoryPressure）。
  • Manager（Operator）
    • 以单实例运行，包含多个 Reconciler：
      • Job Reconciler：将 Job 拆分为 TaskGroup。
      • TaskGroup Reconciler：根据 TaskGroup 的选择器，从 Redis 获取符合条件的空闲机器人，生成对应的 Task（绑定机器人）。这过程就是调度。
      • Task Reconciler：监控 Task 状态。当 Task 被创建且 messageAcked 为 false 时，不直接发送 MQTT，而是依赖 Server 完成下发。Task Reconciler 只需监听任务最终结果（通过消费 robot.events 或定时查询 Redis），更新 Task phase 为 Succeeded/Failed，并处理重试、超时。
      • Robot Reconciler：监听 Robot 状态变化（来自事件或 Redis），更新 Robot CRD 的 conditions（如离线、资源压力）。当 desiredVersion 与 currentVersion 不一致时，创建 OTA Task 下发固件升级指令（同样由 Server 投递）。
  • Server（指令下发组件）
    • 独立部署的轻量服务，单一职责：可靠地将 Task 指令通过 MQTT 下发给机器人。
  • Provision Server(Broker证书签发访问服务器）
    • 独立部署的轻量服务，单一职责：可靠地将机器人需要的Broker访问凭据签发。需要设备使用MQTT连接
  • PostgreSQL
    • 权威元数据存储：机器人、任务、Job 的声明式定义与生命周期状态

组件列表

#

可靠性保证

#

这里的可靠性需要充分考虑如下情况：

• 多个组件忽然宕机
• 机器人忽然大规模消息传递或者离线

这里面实际上诸如kafka, flink, 甚至manager都有本身的可靠性考虑（manager使用微批量（Micro-batching））

整个系统的可靠性是分层解耦、多层防护的：数据平面处理高频流，控制平面处理低频决策，每层都有独立的高可用和恢复机制，故障范围被严格限制。没有任何单点可以导致全局瘫痪或数据丢失。

2.2 数据流

#

包括状态和注册的上行数据流如图，这里注意对于manager和pg的写操作没体现

#

上行（signals + control）：

sequenceDiagram
  participant A as Agent
  participant M as MQTT
  participant PS as Provision Server
  participant KS as Kafka signals
  participant F as Flink
  participant R as Redis
  participant KC as Kafka control
  participant Mgr as Robot Manager
  participant PG as PostgreSQL

  A->>PS: register
  PS->>Mgr: 注册机器人到K8S
  PS->>A: 签发MQTT Broker凭据

  A->>M: heartbeat/task ack/task result
  M->>KS: 写入
  KS->>F: 消费
  F->>R: 更新在线/资源用量

  F->>KC: control event(离线确认 / task 终态 / 资源过载)
  KC->>Mgr: Robot Reconciler 消费
  Mgr->>PG: 更新Robot/Task detailed

  Mgr->>Redis: 更新 Robot Labels相关

下行（Task CRD还需要给出OTA/CONFIG之类的更新内容）：

sequenceDiagram
  participant Mgr as Robot Manager
  participant R as Redis
  participant CRD as CRD(PG)
  participant S as Server
  participant MQTT as MQTT
  participant A as Agent
  participant KS as Kafka signals
  participant F as Flink
  participant KC as Kafka control

  Mgr->>CRD: watch Job spec
  Mgr->>R: 选机 · 匹配 selector
  Mgr->>CRD: 于K8S集群创建 Task PendingDispatch
  S->>CRD: watch Task
  S->>MQTT: publish task/desired
  S->>CRD: Task Dispatched
  MQTT->>A: 投递
  A->>MQTT: ack / result
  MQTT->>KS: 规则写入
  KS->>F: 消费
  F->>R: 更新 task phase
  F->>KC: 创建event：task.acked · task.succeeded
  Mgr->>KC: 消费 task.acked
  Mgr->>KC: 消费 task.succeeded
  Mgr->>CRD: Task Succeeded · Job 汇总

2.3 关键数据定义

#

2.3.1 MQTT Topic 与 Payload

#

连接参数：clean session = false；mTLS 客户端证 CN = robotId；Topic 授权限定每机只能访问自家 {robotId} 前缀。

2.3.2 Kafka：signals 与 control events

#

signals topic（海量，仅 Flink 消费）：

• robot.heartbeat
• robot.task # ack / result
• robot.ota

control events topic（低量，Manager / Server 消费）：

• events.control

partition key 均为 robot_id。

signals Envelope（MQTT 规则 / Server 写入）：

{
  "event_id": "uuid",
  "event_type": "robot.heartbeat",
  "robot_id": "r-001",
  "trace_id": "trace-abc",
  "occurred_at": "2026-06-08T10:00:00Z",
  "schema_version": 1,
  "payload": {}
}

control event 类型（Flink 产出，Manager / Server 消费）：

谁写入 Kafka：

Redis 读模型（Flink 写，Job Reconciler 读）

#

Redis Key（仅 Flink 写入；可从 signals offset 重建）：

Flink 对 signals 的 Redis 投影（按 robot_id keyBy，全量 heartbeat 在此消化）：

Flink 产出 control events 的规则（不进 Redis 重复存储，只发 Kafka）：

CRD 与 status.phase

#

下面给出相关CRD的定义和解释：

Job 是什么：用户提交的业务目标，如“将仓库A的所有货物搬到B”。

包含什么：任务模板（动作类型、参数）、约束（优先级、截止时间）、机器人筛选的 label 选择器（例如 zone=warehouse-a, capability=heavy-lift）。

调度粒度：不直接绑定机器人，只规定“哪些机器人可以做”。

TaskGroup（TG） 是什么：Job 拆分出的一个 调度策略单元，代表“在某个策略下，一批 Task 该如何分派给机器人”。

包含什么：

细化后的选择器（可能从 Job 继承或重新定义，例如 TG1 负责轻载机器人，TG2 负责重型机器人）。

并发策略：最大并发数、是否 Gang Scheduling。

依赖关系：TG 之间的 DAG（例如 TG2 依赖 TG1 完成）。

任务模板：用于生成该组内的 Task。

注意：TG 本身 不存储具体机器人 ID。它只存储“要找什么样机器人”的规则。具体机器人是在 Task 生成时，Manager 结合实时状态（Redis 中的位置、电量、标签）动态选择的。

Task 是什么：一个机器人执行的原子动作，如“移动到 (x,y)”或“抬起货架”。

包含什么：

目标机器人 ID（由 Manager 在生成 Task 时填入）。

具体动作指令（payload）。

生命周期状态（Dispatched → Running → Succeeded/Failed）。

生成时机：Manager 的 TaskGroup Reconciler 根据 TG 的选择器和实时机器人状态，为每个具体动作创建 Task，同时绑定一台机器人。

apiVersion: robot.k8s4r.io/v1alpha1
kind: Task
spec:
  targetRobot: r-001
  command: start_capture
  params: { duration_sec: 600 }
  traceId: trace-abc
status:
  phase: PendingDispatch | Dispatched | Acked | Running | Succeeded | Failed | DeliveryFailed

Job Reconciler 创建 Task 后 不直接调 Server；Server watch PendingDispatch 的 Task 并 MQTT 下发。

PostgreSQL当中的表结构

CREATE TABLE robots (
    robot_id   VARCHAR PRIMARY KEY,
    name       VARCHAR,
    model      VARCHAR,
    labels     JSONB DEFAULT '{}',          -- 静态标签（zone, capability等）
    desired_firmware_version VARCHAR,       -- 期望固件
    desired_config JSONB,                   -- 声明式配置
    version    BIGINT DEFAULT 0,            -- 乐观锁版本号
    online     BOOLEAN DEFAULT false,       -- 由 Manager 根据事件更新
    last_seen  TIMESTAMPTZ,
    current_task_id VARCHAR,
    conditions JSONB DEFAULT '[]',          -- Healthy, CPUOverload 等
    created_at TIMESTAMPTZ DEFAULT now(),
    updated_at TIMESTAMPTZ DEFAULT now()
);

-- Job 表
CREATE TABLE jobs (
    job_id      VARCHAR PRIMARY KEY,
    spec        JSONB NOT NULL,          -- 业务模板、参数、约束
    desired_phase VARCHAR NOT NULL,      -- Pending | Running | Succeeded | Failed | Cancelled
    actual_phase  VARCHAR,
    version     BIGINT DEFAULT 0,
    created_at  TIMESTAMPTZ DEFAULT now(),
    updated_at  TIMESTAMPTZ DEFAULT now()
);

-- TaskGroup 表

CREATE TABLE taskgroups (
    group_id    VARCHAR PRIMARY KEY,
    job_id      VARCHAR NOT NULL REFERENCES jobs(job_id),
    -- 调度策略
    selector    JSONB NOT NULL,   -- 机器人选择器 {zone: "A", capability: "light"}
    strategy    JSONB,           -- 并发控制 {maxConcurrent:5, gangScheduling: false}
    -- 状态
    desired_phase VARCHAR NOT NULL,  -- Pending | Running | Completed | Failed | Paused
    actual_phase  VARCHAR,
    -- 进度统计（由 Manager 定期计算或由 Task 事件驱动更新）
    task_count    INT DEFAULT 0,
    tasks_succeeded INT DEFAULT 0,
    tasks_failed  INT DEFAULT 0,
    version     BIGINT DEFAULT 0,
    created_at  TIMESTAMPTZ DEFAULT now(),
    updated_at  TIMESTAMPTZ DEFAULT now()
);

-- Task 表（之前定义过）
CREATE TABLE tasks (
    task_id      VARCHAR PRIMARY KEY,
    group_id     VARCHAR REFERENCES taskgroups(group_id),
    job_id       VARCHAR NOT NULL REFERENCES jobs(job_id),
    target_robot VARCHAR,
    spec         JSONB NOT NULL,
    desired_phase VARCHAR NOT NULL,
    actual_phase  VARCHAR,
    message_acked BOOLEAN DEFAULT false,
    retry_count  INT DEFAULT 0,
    version      BIGINT DEFAULT 0,
    created_at   TIMESTAMPTZ DEFAULT now(),
    updated_at   TIMESTAMPTZ DEFAULT now()
);

机端 SQLite（任务状态）

#

出站遵循 outbox：先写 SQLite，再 publish MQTT。

2.4 资源模型

#

API 组为 **robot.k8s4r.io/v1alpha1**（版本二对齐为 robot.fleet.io/v1）。资源按职责分为 设备、任务编排、制品下发 三类：

任务编排层级（类比 Deployment → ReplicaSet → Pod）：

Job（用户声明「要做什么」）
  └── TaskGroup（选几台 robot、开几个并发）
        └── Task（执行实例 + MQTT 下发，持久化在 etcd）

调度与下发分离（同一 Task CRD，不同 Reconciler 写不同 phase）：

• Job Reconciler：读 Redis 选机 → 创建 Task（PendingDispatch）→ 收 task.succeeded/failed 写业务终态
• Server：watch Task → MQTT → 写 Dispatched / Acked / DeliveryFailed
• Flink：signals → Redis；关键变迁 → control events

读法要点：

• Robot 管设备身份与 OTA 策略；在线/离线 瞬态在 Redis，持久 status 由 Robot Reconciler 根据 control events 写 CRD。
• Job / TaskGroup / Task 管作业生命周期；Task 是 Manager 与 Server 的边界对象。
• Job / TaskGroup / Task 不承载遥操控制帧（遥操见第三节）。
• OTARollout / ConfigDelivery / Rollout 管制品与配置的分批下发与回滚。

人机与机机 认证分轨：管理员走 OIDC / 自建账号；机器人走 证书 + MQTT 业务注册。

2.4 版本一：Robot Manager + Server

#

删除掉了

2.5 版本二：自建 Fleet 控制面

#

版本二将控制面从 K8s 抽离为 Fleet API + Controller Manager。signals → Flink → Redis / control events 管道与版本一相同；fleet-controller-manager 替代 Robot Manager；边侧 mqtt-gateway 仍 watch Task 并 MQTT 下发。

组件分工

#

借鉴 K8s 的机制

#

PostgreSQL 通用 CR 表（示意）：

resources (
  id, api_version, kind, namespace, name,
  spec JSONB, status JSONB,
  resource_version BIGINT,
  generation BIGINT,
  created_at, updated_at, deleted_at
)

控制面逻辑

#

  fleetctl / Web UI / CI
         │
         ▼
  fleet-api-server (PostgreSQL SoT)
         │ watch
         ▼
  fleet-controller-manager ──读 Redis · 消费 control events
         │
         ▼
  mqtt-gateway ──watch Task · MQTT──▶ Agent

  signals ──▶ Flink ──▶ Redis
              └──▶ control events ──▶ Manager / mqtt-gateway

status 由 Reconciler 写 PG（与版本一写 CRD 对称）；Redis 仍只由 Flink 写。

版本二 API 与版本一 CRD 字段对齐，便于迁移：

apiVersion: robot.fleet.io/v1
kind: Job
metadata:
  name: collect-001
spec:
  taskGroups:
    - name: main
      count: 3
      robotSelector:
        matchLabels:
          site: shanghai
      template:
        command: start_capture
        params: { duration_sec: 600 }

2.6 部分配置细节

#

MQTT Broker:

• Keep Alive：30 秒,时间过短会造成弱网机器人短暂离线缺失状态，过长会导致机器人失连不被感知可能带来风向。
• 持久会话：需要开启持久会话，机器人可能因网络抖动短暂离线，恢复后应当收到离线期间下发的任务。若使用非持久会话，重连时 Broker 会清除所有未消费消息，导致任务丢失
• Session Expiry = 2 小时。覆盖绝大多数网络中断恢复窗口。超过 2 小时可认为机器人长时间失联，应由 Manager 重新分配任务，没必要在 Broker 无限保留消息
• 最大离线消息数 = 1000：防止恶意或故障客户端离线后积压海量消息撑爆 Broker 内存。

KAFKA:

• Hybrid: 磁盘2GB，内存缓冲 64MB：覆盖突发流量高峰（如大量机器人同时上线），相当于缓存约 30 万条心跳消息（每条 200 字节）。一旦写满，溢出到磁盘 2GB，可容纳 1000 万条消息，足以应对 Kafka 短时故障（几分钟至几十分钟）。

• 分区数量：副本数量是按照消费者，就是flink的消费能力决定的，机器人数量估算的10万台机器人，期望每个 Flink 处理 Task 管理的状态量 (≈1000 台)，那么消费者数量为 100（10 万 ÷ 1,000 = 100）。纯读取比处理算子轻量得多，不需要那么高的并行度，设为 50 就足够均衡消费流量，同时不给 Kafka 和 Flink 的网络造成过多小任务

• 基础知识：

  • Broker 数量：就是 Kafka 服务器的物理或虚拟机数量。
  • AZ 数量：可用区（Availability Zone）数量，是云厂商提供的隔离故障域。
  • 副本因子 是 包括 Leader 在内的总副本数。
  • min.insync.replicas 取值：通常设为 副本因子 - 1。例如副本因子=3，设为2，这样允许 1 个副本掉队或宕机，同时仍能保证写入成功且不丢数据。
  • AZ 数量应 ≥ 副本因子数，最好等于副本因子数，以实现最高容灾能力

• 分区Key：按照robot_id进行分区，能够保证机器人的消息顺序，有助于保持flink的状态局部性。

  • 这里有个三选二定律，“三选二”定律就是：在 Kafka 这一层，你没法同时获得：
    • 消息严格顺序（同一台机器人的消息不乱序）
    • 无热点（每条传送带压力均匀）
    • 高并行消费（有很多条传送带且每条都能被人同时处理）
  • 如果出现热点怎么办：
    • 1 在消息进入 Kafka 前（比如在 EMQX 规则引擎里），对热点机器人的消息做处理：给 robot_id 后面加一个随机后缀，比如 bot-A-0、bot-A-1…，这样它的消息就被分散到多个分区，缓解热点。代价是该机器人的消息顺序无法保证，但心跳等数据本就可以容忍短暂乱序（业务端有序列号在 Redis 端兜底）。
    • 如果这个机器人太特殊，直接为它专门建立一个独立的 Topic，单独配置分区数（比如 10 个分区），和其他机器人彻底隔离。这只影响这一个机器人，对其他机器人零影响。
    • 通知机器人降低上报频率，或让它合并数据发送，从源头减少消息量。
  • 如果手动做Kafka 的分区重分配（Partition Reassignment），也就是有新 Broker 加入或需要数据重平衡时
    • afka 的分区重分配对消费者组是透明的。消费者通过 Group Coordinator 得知分区重新分配，会触发 Rebalance。Flink 作为消费者，在 Rebalance 时会自动从 Checkpoint 中记录的 offset 继续消费每个分区。只要 Flink 的 Checkpoint 包含每个分区的 offset，恢复后就能精确接续，不会丢数据也不会重复（基于 EXACTLY_ONCE 语义）
    • 迁移期间的乱序或重复：在 Rebalance 过程中，消费会短暂停止。当消费者重新接管分区时，分区内的顺序不变（分区日志本身未被修改）。因此，不会引入乱序。但是由于 Rebalance 可能导致某些消息在旧消费者上被部分处理但未提交 offset，新消费者会从 Checkpoint 的 offset 开始重放，导致消息重复。这正是你的架构要处理的问题：Flink 写 Redis 是 At-least-once，但业务层通过序列号幂等过滤重复。

• kafka的acks配置是什么？

  • acks 是“写确认的严苛程度”。acks就是自己抄写好了，然后副本也确认了
  • ISR 就是“保持同步的副本集合数量”。min.insync.replicas=2（比如总共 3 个副本，要求至少 2 个抄写员同步），只有他们确认了，才算安全。
  • 配置 acks=all + min.insync.replicas=2 的原因

• 怎么保证的exactly-once：dempotent Producer（幂等生产者）：

  • 每个生产者被分配一个唯一的工号（PID）。它发的每封信给每条传送带（分区）都标上连续编号（序列号）。传送带管理员会记住“这个工号、在这个传送带，最后收到的是第 5 号信”。如果因为网络原因，生产者重发了第 5 号信，管理员一看：“这封我收过”，直接丢掉，单分区内不会重复。
  • 限制：序列号是按“分区+工号”记忆的，跨分区不保证，生产者重启后（新工号）也可能不认识旧的信，所以跨会话不保证。
  • 这个性能影响很低：幂等 Producer 只在上述流程中增加了两步：
    1. 每条消息带一个序列号（一个递增的整数，8 字节）
    2. Broker 端检查序列号：对每个 <PID, 分区> 组合，Broker 在内存中维护最后收到的序列号。新消息到达时，比对序列号——如果是重复的，直接丢弃并返回成功；如果是乱序的（跳号），报错。
  • 请注意，这里要求消费者端配置read_commited，这是消费者端的一个配置 isolation.level，默认值为 read_uncommitted。
    • read_uncommitted：消费者可以立即读取任何已写入分区的消息，不管它们是否属于已提交的事务。这是低延迟的选择，但可能读到后来被中止的事务数据（脏读）。
    • read_committed：消费者只会读取 已提交的事务消息 和 非事务消息。如果消息是由事务生产者写入的，必须等到事务提交后，消费者才能看到。这避免了脏读，实现了事务级别的 Exactly-once 语义。
  • 实测数据（来自 Kafka 官方和社区 benchmark）：
    • 开启幂等 Producer，吞吐量下降不到 3%，延迟增加不到 1ms。
    • 因为 Kafka 本身就是批量发送（linger.ms=10），序列号检查和 ACK 等待都淹没在批处理延迟里了，用户几乎无感。 跨分区、跨重启的 Exactly-Once 需要事务（Transactions）：
  • 生产者开启事务，给一个固定的“事务 ID”（比如 txn-001）。即使重启，因为事务 ID 固定，Broker 能认出是同一个逻辑生产者，延续之前的序列号，跨分区原子写入，并且配合消费者只读“已提交”的信，实现端到端 Exactly-Once。
  • 性能影响很大：需要网络开销，ISR确认，消费者阻塞
  • 实测数据：
    • 事务写入的 P99 延迟可能达到 100ms+（非事务通常在 10ms 内）。
    • 吞吐量：单 Producer 事务吞吐约为非事务的 60-70%。

• kafka是否配置auto.commit控制消费者是否自动向 Kafka 提交已消费消息的 offset。

  • enable.auto.commit=true（默认）：消费者会定期（由 auto.commit.interval.ms 控制，默认 5 秒）自动提交当前已拉取消息的最大 offset。这种方式最简单，但可能导致数据丢失或重复——如果在提交 offset 和实际处理消息之间消费者崩溃，重启后会从上次提交的 offset 继续，可能漏掉一些未处理的消息，或重复处理已处理的消息。
  • enable.auto.commit=false：消费者需要手动提交 offset（调用 commitSync() 或 commitAsync()），通常是在业务逻辑处理成功之后。这样可以精确控制 at-least-once 语义。
  • Flink 作为消费者，不使用 Kafka 的 offset 提交机制。Flink 通过 Checkpoint 将所有算子的状态（包括每个分区的消费 offset）作为一致性快照保存到持久化存储（如 S3）。故障恢复时，Flink 从最近的 Checkpoint 重新加载状态和 offset，实现 Exactly-once 语义。因此，Flink 的 Kafka Consumer 必须配置 enable.auto.commit=false，并且 setCommitOffsetsOnCheckpoints(true

• 多 Consumer 实例的 Partition 分配策略 Kafka 提供了多种分配策略，决定如何将分区映射到 Consumer Group 内的各个消费者：

  策略	特点	适用场景
  RangeAssignor（默认）	按 Topic 分区范围分配，容易导致数据倾斜	简单场景
  RoundRobinAssignor	轮询分配，均匀分布	无特殊顺序要求
  StickyAssignor	尽量保持原有分配，减少分区迁移	减少 Rebalance 开销
  CooperativeStickyAssignor	渐进式 Rebalance，不停止所有消费	生产环境推荐，尤其大规模

  在我们的架构中：Flink 通过 KafkaSourceBuilder 或 FlinkKafkaConsumer 内部管理，它默认使用 StickyAssignor 并配合自己的状态恢复。我们不需要手动指定策略，但应优先选择 CooperativeStickyAssignor 以减少再平衡影响。

Redis：

#

Manager：

#

三、遥操作：实时链路与设备侧 rtele

#

遥操作在产品上 often 需要：多路视频 + 低延迟控制 + VR/手柄输入。这些需求是高于云管平台的，因此实现的时候是基于云管平台的。遥操作平台负责解决谁，遥操作，哪个机器人的问题。简单来说就是遥操作平台做中控，执行的命令包裹好后通过云管下发执行。如图

┌─────────────────────────────────────────┐
│  遥操作中控（会话、RBAC、审计）             │
│  - 谁占用哪台 robot、会话开始/结束          │
│  - WebRTC 视频 / 操控（VolcEngine RTC 等）│
└──────────────────┬──────────────────────┘
                   │ 封装 API（不浏览器直连 K8s）
                   ▼
┌─────────────────────────────────────────┐
│  K8S4R：Robot 在线 + Job 启停 teleop 栈   │
└──────────────────┬──────────────────────┘
                   │ MQTT
                   ▼
              机器人 Agent + rtele

允许遥操 建议同时满足：

1. 云管平台Robot 在线（及策略要求的标签）；
2. 中控侧 会话锁 未被他人占用。

设备状态以云管为准；占用以中控数据库为准——避免两套逻辑长期打架。

3.1 设备侧任务语义

#

设备侧任务实际上是我写的rtele基于 火山引擎 VeRTC 的远程遥操作程序，负责机端（或边端）,接入云：

目前实际上是直接VR redirect到星海图机器人上面，协议包含了位资，pose等数据

典型数据流（简化）：

设计要点：

• 低延迟控制不要塞进云管的任务里里每帧更新；云管最多负责 Task 启动 rtele。
• 视频编码在边缘完成（如 Rockchip MPP或者其它硬编码），减少上行带宽。
• 遥操作产生的 原始记录（MCAP 等）是数据平台的输入，不是 Fleet 的持久化职责。

我使用5G CPE测试的过程中发现，实际上纯网络视频传输的延迟150-200ms，网络好的时候100ms。主要的卡顿的点拆分是两种

• 操作卡顿：操作卡顿的问题在于控制消息的发送端和接收端的频率并不相同，控制段频率可能文档100hz，接收端可能在30到200hz跳变。因此，我最终在设备端加了一个prediction，使用线性速度和角速度预测
• 视频卡顿：视频卡顿的来源主要是CPU对视频编解码上，发送端和接收端都必须使用硬编码才能解决。当时测试的时候我做到了操作延迟可接受（100-400ms），视频延迟可耻的到了1s，就是软件编解码的问题（机器人侧jpeg转h264，接收端h264解码）

四、数据平台

#

数据平台要解决的是另外一类问题：大规模数据采集、元信息管理、自动化流水线、信息汇总和展示、输出到下游。一些人可能好奇，设计这东西干什么，有什么难得？在创新中心，采集数据的量级是百万小时数据，一小时数据差不多 10G，百万小时的数据就是 10,000TB，也就是 10PB。这种量级下的数据管理、处理、回放、质量检查、训练集发布都是问题，因此这里的设计目标是对标创新中心的数据管理现状。

数据平台的核心业务不是“把文件存起来”，而是把采集产生的原始数据组织成可查询、可复现、可训练、可评测的 Dataset。所以它的主线应该是：

UploadSession -> RawFile -> Episode -> [自动预标/后处理] -> Sample/Clip -> [人工标注 + 质检] -> Dataset -> [导出训练格式] -> EvalRun / TrainRun -> ModelArtifact

• UploadSession：一次上传事务，记录 upload_id、robot_id、task_id、文件列表、hash、大小、schema_version、上传状态。
• RawFile：原始文件对象，例如 MCAP、视频、图片、点云、日志。RawFile 只追加，不直接修改。
• Episode：一次采集、遥操或任务执行的连续片段，绑定机器人、任务、操作员、时间范围、固件版本、标定版本、Agent 版本等上下文。
• 自动预标 / 后处理（GPU）：用模型对 Episode/Sample 做自动打标与预处理，例如 VLM 生成语言描述、检测/分割、pose 估计、自动切分技能段、算 embedding。产物是"预标注"，给人工校正打底，也可直接作为弱标注。这是 GPU 作业，不是 Spark。
• Sample/Clip：从 Episode 切出的训练样本或评测片段，例如 10 秒轨迹、多帧图像、点云窗口、动作序列。
• 标注（人在环）：在自动预标基础上，由 Label Studio/CVAT 或外包团队做人工校正，产出语言指令、任务分段、成功/失败、检测/分割/关键点、质量等标注。标注可版本化，是 Dataset 的输入。
• 质检（自动 + 人工）：数据质检与标注质检两类，自动规则/模型打分 + 人工抽检，结论写回 sample.quality_score，是 Dataset freeze 的门槛。
• Dataset：冻结后的样本集合，包含版本、样本清单、过滤条件、split、hash、血缘和质量报告。训练和评测只能引用已发布 Dataset。
• 导出（训练格式）：Dataset 只是引用清单，dataloader 喂不进清单，所以再把命中样本物化成 LeRobot / RLDS / webdataset / HDF5 的 shard 落 OSS，登记为 dataset_export。
• EvalRun / TrainRun：评测任务和训练任务，记录代码版本、参数、输入 Dataset 版本、输出指标和产物。跑在 GPU（Ray/PyTorch）上，不在 Spark 上。
• ModelArtifact：模型文件、评测报告、导出格式、部署版本，反向引用 TrainRun 和 Dataset。

这样设计之后，数据平台回答的问题就从“文件在哪”变成了“这个模型用了哪些机器人、哪些任务、哪些数据版本、哪些标注和过滤规则训练出来”。

4.1 架构设计

#

数据平台的编排我先用阿里云托管 serverless（EventBridge + FnF）做了起步版（见 4.1.1 / 4.1.2），但随着需求扩到自动预标、人工标注、质检、GPU 训练这些异构作业，托管方案在"统一编排异构算子"上开始别扭，最终演进到自建的 Dagster + Spark/DuckDB/Ray + K8s 架构（见 4.1.3，选型对比与可靠性评估见 4.1.4），并顺手砍掉了 EventBridge。

不管哪一版，先立一个四层心智模型，后面所有组件往这四层里对号入座就不乱了：

一句话：触发层叫醒编排层，编排层决定让哪个执行引擎干、干完再让谁干，执行引擎把结果写进存储层。 注意 Kafka 不在"触发层"里——它是业务事实日志（记录发生过什么、可重放），是账本不是门铃，触发/通知不要塞进 Kafka。

初期不承诺实时性，T+1 是基本目标。下面先给托管 serverless 的两版起步实现，再给当前自建统一版。

4.1.1 托管 serverless 起步版（简单实现）

#

这是最初的托管版：编排用 FnF、触发用 EventBridge。作为演进历史保留，当前方向见 4.1.3。

简单实现如下所示

graph TD
    subgraph 数据源
        D1[IoT/业务系统]
        D2[文件上传客户端]
    end

    subgraph 接入层
        GW[上传网关<br/>API Gateway + FC<br/>鉴权/校验/路由/签发上传URL]
        OSS_RAW[OSS Raw Zone]
        KAFKA[Kafka<br/>业务事件日志<br/>upload/file/episode/dataset]
    end

    subgraph 大脑中枢
        EB[EventBridge<br/>OSS事件/定时T+1/人工触发/告警]
        FNF[Serverless工作流 FnF<br/>编排DAG]
        STATUS[(Tablestore<br/>UploadSession/IngestJob/Dataset状态)]
    end

    subgraph 处理与存储
        SPARK[EMR Serverless Spark<br/>Batch / 未来微批]
        subgraph 数据湖分层
            OSS_QUARANTINE[OSS Quarantine Zone]
            OSS_BRONZE[OSS Bronze Zone]
            OSS_SILVER[OSS Silver Zone]
            OSS_GOLD[OSS Gold Zone]
            OSS_DATASET[OSS Dataset Zone]
            OSS_ARTIFACT[OSS Artifact Zone]
        end
        ICEBERG[Iceberg表 + DLF REST Catalog]
        LANCE[(Lance数据集)]
    end

    subgraph 查询与服务
        TRINO[Trino/Spark SQL<br/>交互查询]
        BI[QuickBI]
        API_QUERY[数据服务API]
    end

    subgraph 控制面
        IAC[Terraform IaC<br/>网络/存储/计算/权限/日志]
        CICD_INFRA[CI/CD 基础设施流水线]
        CICD_APP[CI/CD 应用/作业发布流水线]
        CONFIG[配置中心<br/>OSS+FC<br/>版本/灰度/回滚/缓存]
    end

    %% 数据流
    D1 -->|HTTP小JSON| GW
    D2 -->|请求上传URL| GW
    GW -->|签发上传URL| D2
    D2 -->|直传文件| OSS_RAW
    OSS_RAW -->|ObjectCreated外部事件| EB
    GW -->|业务事实事件| KAFKA
    EB -->|外部触发| FNF
    FNF -->|提交作业| SPARK
    SPARK -->|读业务事件| KAFKA
    SPARK -->|异常数据| OSS_QUARANTINE
    SPARK -->|处理| OSS_BRONZE
    SPARK -->|处理| OSS_SILVER
    SPARK -->|处理| OSS_GOLD
    SPARK -->|发布训练集| OSS_DATASET
    SPARK -->|保存训练/评测产物| OSS_ARTIFACT
    SPARK -->|更新数据平台状态| STATUS
    SPARK -->|读写| ICEBERG
    SPARK -->|写Lance元数据| ICEBERG
    SPARK -->|写Lance文件| LANCE
    TRINO --> ICEBERG
    API_QUERY --> TRINO
    API_QUERY --> STATUS
    BI --> TRINO

    %% 控制流
    IAC -->|创建/管理| 全部
    CONFIG --> GW
    CONFIG --> SPARK
    CICD_INFRA -->|Plan/Apply| IAC
    CICD_APP -->|发布代码| GW
    CICD_APP -->|发布作业| SPARK

4.1.2 托管 serverless 完整版

#

graph TD
    subgraph 数据源
        D1[IoT设备/业务系统]
        D2[文件上传]
    end

    subgraph 数据面 - 接入层
        GW[上传网关<br/>API网关+FC]
        KAFKA[消息队列Kafka<br/>upload / file / episode / dataset事件]
        OSS_RAW[OSS原始文件]
    end

    subgraph 控制面 - 大脑中枢
        EB[事件总线 EventBridge<br/>OSS事件/定时/人工/告警触发]
        FNF[Serverless工作流 FnF<br/>决策与状态管理]
        STATUS[(状态存储<br/>Tablestore/RDS<br/>业务状态/运行状态)]
    end

    subgraph 数据面 - 处理与存储
        SPARK[EMR Serverless Spark<br/>ETL / 清洗 / 打宽 / 评测]
        ICEBERG[(OSS + Iceberg<br/>结构化数据湖)]
        LANCE[(Lance数据集<br/>非结构化/向量)]
        TABLESTORE[(Tablestore<br/>数据平台状态/修正缓冲)]
        DLA[DLA/Trino<br/>交互式查询引擎]
    end

    subgraph 数据面 - 服务与查询
        API_QUERY[数据服务API]
        BI[QuickBI / 看板]
    end

    subgraph 控制面 - 治理与运维
        DLF[DLF数据湖构建<br/>元数据目录]
        CONFIG[配置中心<br/>OSS+FC]
        MONITOR[云监控 + SLS<br/>日志/指标/告警]
        AUDIT[ActionTrail审计]
        IAM[RAM权限 + KMS]
        IAC[Terraform IaC]
        CICD[CI/CD 云效/GitHub Actions]
    end

    %% 数据流
    D1 -->|HTTP上报| GW
    D2 -->|直接上传| OSS_RAW
    D2 -->|元数据| GW
    GW -->|业务事实事件| KAFKA
    GW -->|文件先存OSS| OSS_RAW
    OSS_RAW -->|ObjectCreated 事件| EB
    EB -->|定时事件| FNF
    EB -->|外部触发| FNF
    FNF -->|发起Spark作业| SPARK
    FNF -->|调用轻量函数| GW
    SPARK -->|读取业务事件| KAFKA
    SPARK -->|读写| ICEBERG
    SPARK -->|读写| LANCE
    SPARK -->|写入数据平台状态| TABLESTORE
    SPARK -->|更新作业状态| STATUS
    FNF -.->|查询/更新状态| STATUS
    DLA -->|SQL查询| ICEBERG
    API_QUERY --> DLA
    API_QUERY --> TABLESTORE
    BI --> DLA
    BI --> TABLESTORE

    %% 控制流
    DLF -->|元数据服务| ICEBERG
    DLF -->|元数据服务| DLA
    CONFIG -->|配置下发| GW
    CONFIG -->|配置下发| SPARK
    MONITOR -->|采集日志指标| GW
    MONITOR -->|采集日志指标| SPARK
    MONITOR -->|采集日志指标| FNF
    AUDIT -->|API调用记录| IAM
    IAM -->|权限控制| GW
    IAM -->|权限控制| SPARK
    IAM -->|权限控制| DLA
    IAC -->|资源编排| 全部
    CICD -->|自动化部署| IAC

4.1.3 当前方向：统一编排 + 三档执行引擎（Spark / DuckDB / Ray）

#

起步版的问题是：编排器（FnF）只擅长提交 Spark，而数据平台真正要统一调度的是多种形态完全不同的作业——大批量 ETL、小批量轻处理、GPU 作业（自动预标/后处理/训练/评测）、人工任务（标注/质检）。所以把编排层换成 Dagster（与 Airflow 3 的对比和可靠性评估见 4.1.4），执行层分成三档引擎，触发层收敛进编排器自身，EventBridge 去掉。

编排层为什么是 Dagster： 它是 asset 优先的编排器——不围绕"任务"建模，而是围绕"数据资产"建模：每张 Iceberg 表、每个 Dataset 在 Dagster 里是一个 asset，声明"依赖谁、怎么生产"；平台自动获得版本（code_version/data_version）、血缘图、新鲜度检查、asset check 质量门。它同样是"被定时或事件叫醒 → 物化一串有依赖的资产 → 就睡下"的调度器，自己一行计算都不做，只发号施令。

执行层三档引擎，按量路由： 不是什么活都值得起一个 Spark 集群。执行引擎分三档，都被编排器调度，区别只在"什么活给谁"：

记法：结构化大批量、SQL、写湖 → Spark；小批量轻活 → DuckDB 单容器（DuckDB 读 Iceberg 已成熟，小批量写用 pyiceberg 或轻量 Spark 兜底）；Python、GPU、模型、多模态 → Ray。 一条资产管道里可以前几步 Spark、中间几步 Ray。关键设计是任务声明和执行引擎解耦：编排器里一个 asset 只声明输入/输出/资源需求，由路由规则（本身就是热配置，见 4.3.5）决定派给哪档引擎——80% 的小批次任务用 DuckDB 跑，成本和延迟都降一个量级。三档都跑在 K8s 上，GPU 作业由 Volcano/Kueue 做 gang scheduling、配额与抢占。

GPU 是一等公民：不再像起步版那样把训练划到平台外。GPU 在三处都用——自动预标/后处理、自动质检、训练/评测——统一通过 Dagster 提交到 K8s+Volcano。

触发层为什么能砍掉 EventBridge：Dagster 自带三种触发，正好覆盖数据平台的全部触发需求；T+1 场景对延迟不敏感，不需要云原生即时事件推送：

人在环不需要新总线：标注/质检要等人干几天，但 Dagster 不会挂着干等——把流程切成"派活"和"收活"两段短 job，中间几天 Dagster 根本不在场，长期状态存 Tablestore；“人干完了"就是一个调 Dagster API 的外部事件（见 4.2.5）。

当前版组件表（替换起步版的 EventBridge/FnF 行）：

当前版数据流：

graph TD
    subgraph 触发层
        CRON[定时 schedule T+1]
        API[GraphQL API<br/>人工完成/CI/网关回调]
        SENSOR[sensor 轮询 OSS/状态]
    end
    subgraph 编排层
        AF[Dagster<br/>只指挥：资产依赖/重试/backfill/血缘/UI]
    end
    subgraph 执行层
        SPARK[Spark on K8s<br/>结构化 ETL / freeze]
        RAY[Ray on K8s<br/>预标/后处理/训练/评测]
    end
    subgraph 存储层
        OSS[(OSS)]
        ICE[(Iceberg)]
        LANCE[(Lance)]
        TS[(Tablestore 状态)]
    end
    subgraph 人工
        ANNO[Label Studio / CVAT / 外包]
    end
    CRON --> AF
    API --> AF
    SENSOR --> AF
    AF -->|提交批作业| SPARK
    AF -->|提交GPU作业| RAY
    AF -->|建任务/通知| ANNO
    ANNO -.->|标完回调| API
    SPARK --> ICE
    SPARK --> OSS
    RAY --> LANCE
    RAY --> ICE
    AF -->|更新流程状态| TS

4.1.4 编排器选型：为什么最终选 Dagster（对比 Airflow 3）

#

编排层有两个认真的候选，最终选了 Dagster。对比以 2026 年现状为准：

Airflow 3（3.0 于 2025 GA，现 3.2）补齐了老版本的几个大短板：事件驱动调度（AssetWatcher 监听外部事件源，Kafka/SQS 开箱支持，可自定义 trigger 适配数据库等其它事件源；也支持 push 式 REST API 上报资产事件）、DAG versioning 内置、asset 分区。但它不 serverless：自托管要 API server + scheduler + DAG processor + triggerer + workers + 元数据库，规模化后普遍估算要 0.5–1 个平台工程师投入；AWS 有 MWAA Serverless，国内云上基本等于自托管。

Dagster 是 asset 优先的编排器：数据资产是一等公民，“每份数据带版本、批次、血缘、新鲜度、质量检查"是它的原生概念——software-defined assets、code_version/data_version、asset checks、freshness check、内置血缘 UI。sensor 天然适配"数据库变更/外部事件/定时"三类触发；Components + YAML 声明式（加新数据源 = 加一个 YAML）正好是配置驱动架构的官方范式。自托管比 Airflow 3 轻（webserver + daemon + PG + code location）。托管版 Dagster+ 在海外云，国内同样自托管。

开源版与可靠性评估（选型前必须过的一关）：

• OSS 是真开源不是阉割版：Apache 2.0，GitHub 15.8k stars、410 贡献者、418 个 release（2026 年仍高频发版）。asset 模型、sensor、调度、分区、backfill、asset checks、freshness、血缘 UI、Components 全在开源版。
• 商业版（Dagster+）圈走的是企业运维面：托管 hosting、UI 级 RBAC/SSO、审计日志、告警面板。对本平台影响最大的两点：OSS 的 UI 没有 RBAC（需网络隔离或前置网关兜底）；告警自己接（sensor/hook 打钉钉，正是 4.3.7 的设计，不算额外负担）。
• 诚实的风险：Dagster Labs 共融资约 4900 万美元，最后一轮是 2023 年的 Series B，且项目是单一公司主导——不像 Airflow 有 Apache 基金会和多厂商共同维护。公司若转向，OSS 会失速。兜底是 Apache 2.0 可以继续用/fork。
• 真正的风险对冲不在选型，在架构：编排器只做控制面、计算全在 Spark/Ray/DuckDB 作业镜像里，意味着编排器里没有业务逻辑本体。万一将来要换掉 Dagster，迁移的只是"胶水层”，不是重写平台。选型风险被这条原则封在了可控范围内。

结论：选 Dagster。 本平台的核心诉求就是"资产 + 版本 + 血缘 + 配置驱动”，这是 Dagster 的主场；它不胜出的两点——生态广度和基金会治理——对"执行体只有 Spark/DuckDB/Ray 三种"的平台不构成实质影响。

不管选谁，有一条减负原则都成立：编排器只做控制面，绝不在自己进程里跑数据任务——所有计算都提交到 K8s。这样编排器本体就是两三个 pod + 一个 PG，“自己维护集群很麻烦"的担忧并不成立，重的部分仍然弹性伸缩。后文按 Dagster 行文；若读者用 Airflow，概念一一对应（job→DAG、sensor→Sensor、GraphQL API 触发→REST API 触发）。

4.2 详细设计

#

上述架构图组件很多，本节按三步把数据流如何组织到架构里讲清楚：

• 4.2.1 数据的逻辑结构：数据分了哪几层、每层在 Iceberg 里对应什么表、表结构长什么样、存在哪。
• 4.2.2 数据的业务逻辑结构：Episode、Dataset、训练/评测这些业务实体是什么，它们怎么引用底层数据，数据和业务在哪一层接上。
• 4.2.3 业务流如何运行：从上传入湖、到冻结 Dataset、到训练，整条流程分别由谁触发、怎么跑。

4.2.1 数据的逻辑结构

#

Iceberg 中有两类表：

• 分层数据表：才有 Medallion 层级，是真正的加工数据，从 Bronze→Silver→Gold 越来越精炼。比如 bronze_imu、silver_imu、gold_*。
• 索引/目录表：描述“数据有什么、怎么关联”的账本，体积小，贯穿所有层、不属于某一层。比如 raw_file、episode、episode_file、sample。

4.2.1.1 分层数据表，供分析

#

先把整条流水线的阶段说清楚——摄取 → 统一格式 → 标准化加工 → 入仓 → 消费/激活，每个阶段对应一个 Medallion 层：

机器人场景下各层的例子和可变性：

这套分层在物理上各司其职：

• 文件本体 → OSS，按 raw/、bronze/、silver/、gold/ 目录分层（桶与路径规划见 4.3.2）。
• 结构化数据和它们之间的关系 → Iceberg 表（表文件也落在 OSS，DLF 统一管 Catalog），用 Spark/Trino 跑 SQL。
• 多模态样本本体（图像、点云、embedding） → Lance，被 Iceberg 表用指针引用。
• 运行态点查（upload/作业状态） → Tablestore。
• 在线消费 → 需要点查/亚秒响应的结果表，导出到 Tablestore 或 OLAP（StarRocks/ClickHouse），在线 API 绝不直查数据湖。

更具体来说就是：

4.2.1.2 规范信号模型与映射配置：Bronze 怎么吃下异构数据

#

现实是 Raw 什么格式都有（MCAP、rosbag2、HDF5、mp4、自研二进制），MCAP 里的 topic 命名和消息 schema 还随机型/固件版本各不相同，而且业务侧灵活多变。解法是把"异构"拆成两个正交的问题，分别用插件和配置吸收：

第一层：容器格式（怎么打开文件）→ parser 插件注册表

Bronze 的入口不是"一个 MCAP 解析器”，而是一个 parser 注册表：每种容器格式一个插件，职责统一——把容器打开成流 parser(file) → 迭代器[(stream_name, ts, payload)]。raw_file.format → 用哪个 parser 是一条热配置路由。新格式 = 注册一个新 parser 插件，平台代码不动。

第二层：语义（topic 是什么信号）→ 规范信号模型 + 映射配置

A 机器人叫 /cam0/image，B 机器人叫 /camera/front/image_raw，其实都是"前置相机图像"。平台定义一组规范信号类型（canonical signals）——camera_image、imu、joint_state、ee_pose、action_cmd……Bronze 表按规范信号建（这就是 4.2.1.1 里 bronze_<signal> 的由来），每个机型/固件版本配一份 topic→信号 的映射配置：

{
  "robot_model": "X1", "fw_range": ">=2.3", "config_ver": "v14",
  "mappings": [
    { "topic": "/cam0/image", "signal": "camera_image", "params": { "camera_id": "front" } },
    { "topic": "/imu/data",   "signal": "imu",
      "transforms": [ { "type": "scale", "field": "accel", "factor": 9.81 } ],
      "expect": { "freq_hz": [180, 220] } }
  ]
}

映射表里没有的 topic 进 bronze_unmapped 登记 + 告警提示补映射，不阻塞已知信号入湖——和 Quarantine 一个思路。expect 是顺手内嵌的质量契约（预期频率、取值范围），入湖时自动变成 DQ 检查。

映射是业务自助的，但要划清配置能表达什么的边界（否则重演"配置图灵完备"的灾难，见 4.3.5）：

业务自助的成败不在 JSON 格式，在四道护栏：

1. Schema 校验：提交时按 JSON Schema 验格式。
2. Dry-run 预览（最关键）：提交映射时拿真实样本文件跑一遍，返回覆盖率报告——哪些 topic 命中、哪些落 unmapped、字段类型对不对、频率是否在 expect 内。业务看着报告确认才允许发布。
3. 版本 + 审批 + 灰度：config_ver 递增、轻量审批、先对一个批次生效再全量（复用 4.3.5 框架）。
4. 可追溯可重跑：config_ver 在批次五元组里（4.3.1），映射写错的批次能精确圈出、改完按 batch_id 重刷。

另外三个配套设计：

• 两级治理：“有哪些规范信号、各含什么字段”（信号本体）是平台资产，慢变——加新信号类型 = 新建 bronze 表，走工程评审；“哪个 topic 对到哪个信号"是业务资产，快变，完全自助。两级分开，业务灵活性才不会腐蚀 Bronze 的 schema 稳定性。
• 自动生成映射草稿：MCAP 是自描述的（内嵌 topic 列表和 schema），第一个文件上传后平台自动 introspect，按命名惯例预生成草稿（*/imu* → imu），业务只需确认修补，不从空白写起。配合 bronze_unmapped 报表，闭环是：新 topic 出现 → 报表提示 → 补一行映射 → 下一批自动生效。
• 存哪：配置表为 SoT（版本化、热加载），镜像一份进 Iceberg ops 便于 SQL 分析；写入口双轨——业务走 UI 表单，工程走 git PR + CI 校验，落点相同。这个"声明生成资产"的模式正是 Dagster Components 的官方玩法。

4.2.1.3 打散与重组：业务要的不是散表，怎么拼回来

#

分层把数据打散成一张张信号表，但真实业务的消费单位从来不是散表——训练要的是时间对齐的多模态整块。这不是设计矛盾，而是一个沙漏结构：打散是为了治理，重组是为了消费；散表是稳定的中间资产，业务形状是从它派生的廉价投影。

Raw（整块文件）
   ↓ 打散：按信号规范化，为了质检/对齐/schema 演进（治理动作只能按信号做）
Silver（散表，中立，稳定资产）
   ↓ 重组：按消费者拼形状，廉价、可重建、可废弃
Gold 宽表 / Lance bundle / export / API 文档（业务形状，可以有 N 个）

Gold 不是一个大表，是一组面向消费者的表——一个业务问题一张表。“打宽"有两种，形状完全不同：

打宽之一：实体级宽表（把 join 提前做掉）。分析师看"每个 episode 的概况"本来要 join 五张表，宽表提前物化，每行一个 episode：

gold_episode_wide:
episode_id | robot_id | scene   | duration_s | success | avg_quality | has_lang_anno | ...
ep-001     | r-01     | kitchen | 143        | true    | 0.92        | true          | ...

打宽之二：时间步级对齐（训练真正吃的形状）。各信号在 Silver 里各自一表、各自频率（相机 30Hz、关节 500Hz、指令 100Hz），重采样到统一时钟后每行一个时间步、列是所有信号：

episode_id | t     | image_uri   | joint_pos[14] | ee_pose[7] | action[14]
ep-001     | 0.00s | lance://... | [...]         | [...]      | [...]
ep-001     | 0.02s | lance://... | [...]         | [...]      | [...]

这个形状物化进 Lance（列式、训练随机读高效），它就是 sample 的本体——所以训练消费的从来不是 silver_imu 这些散表，散表在这条路上是治理和血缘的账本。

“跟着业务走"的应对：让业务形状成为可抛弃的派生物。 成本结构是不对称的：Silver 是重资产（清洗复杂、重刷昂贵），但按信号中立建模，业务怎么变它都不动；Gold/export 是轻资产，全部从 Silver 派生、可重建。加一个新业务 = 加一个下游 asset（一张 Gold 表或一个导出格式），Dagster 血缘图上一眼看清它依赖哪些信号；业务黄了就废弃这个 asset，Silver 毫发无损。且不是每个业务形状都要物化：高频/复杂 join 的（训练包、核心宽表）物化，探索性低频的用 Iceberg view / Trino 现拼，被证明高频了再物化。

粒度也是参数，不是架构约束。 sample 的真实定义是"训练/评测的消费单位”，粒度由 slicer_version 承载：

导出同理：LeRobot 格式本来就是"一个 episode 一组文件”；甚至 export = 按清单把原始文件拷出 + index 也合法。架构只保证一件事：导出严格按 dataset_sample 清单物化；片段还是整文件，是 slicer 和 exporter 的参数。

两个反模式，划好红线：

• 不要让 Silver 迁就业务：一旦某业务说服你在 Silver 里加"它要的形状”，Silver 就腐化成第二个 Gold，所有下游被这个业务绑架。业务耦合只允许存在于 Gold 及以下。
• 不要绕过分层直取 Raw：短期少一跳，长期每个业务自己维护一套"从 MCAP 解析开始"的私有管道，质检和血缘全部失效——那就是数据平台存在之前的世界。

4.2.1.4 索引/目录表

#

索引/目录表记录数据实体及其关联关系，体积小、贯穿各层、不归属于某个 Medallion 层。核心表结构如下。

raw_file——原始文件登记表，每个上传文件一行：

episode——一次连续采集/遥操的语义单元：

Episode 边界按优先级递减的三种来源确定，由 source 字段标记：

• declared（首选）：采集端最清楚一次任务的起止，故 manifest 直接带上 episode_id，上传时 raw_file 即归属确定，平台无需猜测。
• auto（兜底）：源头未带 episode_id 时，由 Episode 生成作业扫描同一 robot_id + task_id 的 raw_file，按时间间隙规则聚合，规则随 generation_rule_ver 版本化。
• manual（例外）：自动切错时，通过 merge/split/relabel 人工校正，并记为 override 以便审计。

episode_file——episode 与 raw_file 的多对多关联表，使 episode 可跨多个文件或多次上传：

ordinal 由文件 start_ts 排序确定，生成作业按 (episode_id, file_uri) 幂等写入，保证重刷后次序稳定、不乱序。

sample——从 episode 切出的训练/评测样本，本体物化在 Lance：

quality_score 是综合分，初期可只用它过滤；但质量本身是多维的（时间连续性、传感器异常、动作合理性各有分数），故预留 quality_tags / quality_detail，后续支持按维度筛选而不必改表。

4.2.1.5 实体关系与多对多的处理

#

把上面几张表连起来，实体关系如下（含跨到业务层的 dataset）：

erDiagram
    raw_file       ||--o{ episode_file   : "出现在"
    episode        ||--o{ episode_file   : "包含"
    episode        ||--o{ sample         : "切出"
    episode        ||--o{ annotation     : "被标注"
    sample         ||--o{ annotation     : "被标注"
    episode        ||--o{ entity_tag     : "被打 tag"
    sample         ||--o{ entity_tag     : "被打 tag"
    sample         ||--o{ dataset_sample : "被引用"
    dataset        ||--o{ dataset_sample : "包含"

各关系的基数与实现方式：

多对多怎么落地：用关联表，不在主表里塞数组。 raw_file 和 episode、sample 和 dataset 都是多对多，统一用一张只含两个外键的关联表（episode_file / dataset_sample）表达。好处是可双向 SQL join、可加附加列（如 ordinal），且写入幂等。

处理（写入）关系：

• episode：入湖作业按 episode_id 做 MERGE upsert——新 id 则 INSERT，已存在则 UPDATE 扩展时间边界（start_ts=least(...)、end_ts=greatest(...)）。因此同一 episode 跨多次上传也只有一行，且幂等。

MERGE INTO episode t USING staged s ON t.episode_id = s.episode_id
WHEN MATCHED THEN UPDATE SET t.start_ts = least(t.start_ts, s.start_ts),
                             t.end_ts   = greatest(t.end_ts, s.end_ts)
WHEN NOT MATCHED THEN INSERT (episode_id, robot_id, task_id, start_ts, end_ts, source) 
                      VALUES (s.episode_id, s.robot_id, s.task_id, s.start_ts, s.end_ts, 'declared');

• episode_file：按 (episode_id, file_uri) 做 MERGE，ordinal 由 raw_file.start_ts 升序生成；重刷不产生重复行、次序稳定。
• sample：由切片作业从 episode 的 silver 数据切出，按 slicer_version append；换切法只是追加新 slicer_version 的样本，老样本不动，多套并存。
• dataset_sample：冻结 Dataset 时一次性写入命中的 (dataset_version, sample_id)，之后只读不改。

整条链 raw_file ──(episode_file)── episode ──< sample ──(dataset_sample)── dataset，把对象存储中的文件组织成可跨文件 SQL 查询、可被业务层引用、可反查血缘的结构化记录。数据层到 sample 为止，其上即业务层（见 4.2.2）。

4.2.1.6 标注与质检表

#

标注和质检是数据层到 Dataset 之间的关键一环。标注横跨两个粒度：episode 级（语言指令、任务分段、成功/失败、质量）体积小，直接进 Iceberg；帧级（检测框、分割掩码、关键点）体积大，本体落 OSS/Lance，Iceberg 只存指针——和 sample.lance_uri 一个套路。质检本质也是"对某个对象打标"，复用同一套框架。

annotation_task——一个下发给标注工具/外包的批次：

annotation——标注结果，可版本化，既是数据也是血缘：

qc_result——质检结论，写回 sample.quality_score 的依据：

有了这两张表，sample.quality_score 就有了明确的来源（auto-QC + 人工抽检），而不再是一个凭空出现的字段。

4.2.1.7 Tag 与自定义组织

#

episode/sample 的固定字段（robot/task/固件版本）覆盖不了自由组织维度——“厨房场景"“夜间采集"“这批标定有问题"这类随业务不断新增的切面，需要一个 tag 系统。先和 annotation 划清界限：

两者有交集：VLM 场景分类既是弱标注也是筛选维度——这种"存一份在 annotation，投影一份成 tag”。

建模沿用 4.2.1.5 的原则（多对多用关联表，不在主表塞数组）：

entity_tag——tag 关联表，可打在任何实体上，按 (target_type, target_id, tag_key) MERGE 幂等：

tag_def——词表治理：tag_key, allowed_values, owner, description。没有它，三个月后就是 kitchen/Kitchen/厨房 三种写法的垃圾场；词表本身是热配置（见 4.3.5），加新 tag 维度不改代码。

tag 在各流程的体现点：

1. 上传入湖：manifest 直接带声明 tag（scene=kitchen），入湖时写入。
2. Silver 加工：规则自动打（夜间/时长/传感器组合），DuckDB/Spark 顺手干。
3. 预标（Ray）：模型打标（场景分类、动作类型），投影成 source=model 的 tag。
4. 人工圈选：数据浏览 UI 打 tag → 先写 Tablestore 缓冲（复用"操作员修正暂存”）→ 小批量作业定期 MERGE 进 Iceberg，人每点一下不直接动湖。
5. Dataset freeze：filter_expr join entity_tag 筛选——tag 是圈数据的主要维度。tag 可变不破坏可复现性，因为权威是 dataset_sample 命中清单（4.2.2 的原则正好兜住）。

“自定义如何组织"用三级组织模型回答：

探索时用 view 圈，圈好了 freeze 成 dataset，三级正好衔接主线；saved view 本质是一条配置（存配置表），不是新组件。

4.2.2 数据的业务逻辑结构

#

4.2.1 的数据层加工到 sample 就停了——sample 只是“一条条可用的样本”，还不是业务。业务层在数据层之上，回答“这次训练/评测到底用了哪批样本、能不能复现”，核心业务实体是 Dataset、TrainRun/EvalRun、ModelArtifact。它们同样落成 Iceberg 表：

**Gold 和 Dataset 不重合，**职责完全不同：

一句话：Gold 是“表”，会变；Dataset 是“某次训练的物料清单”，不变。 Dataset 不复制数据，只用 dataset_sample 记录引用了哪些 sample_id + 版本 + hash + split。

关于 filter_expr 的可复现性需要注意：过滤表达式可能引用 UDF 或外部函数，环境变化后未必能再解析。因此 冻结 Dataset 时以 dataset_sample 命中清单为权威重放依据，filter_expr 与 code_ver 仅用于审计和展示，不用于重放。这样即使过滤逻辑的环境失效，已发布 Dataset 仍可凭固定的 sample_id 清单精确复现。

数据层和业务层通过 sample 连接。

Dataset 只是引用清单，训练要的是物化的格式包。 freeze 出的 dataset + dataset_sample 是不可变的"物料清单”，但 PyTorch dataloader 喂不进"清单”。所以再加一步 dataset_export：由一个作业按清单把样本物化成 LeRobot / RLDS / webdataset / HDF5 的 shard，落 OSS，元数据登记进 dataset_export。同一个 Dataset 版本可导出多种格式喂不同框架，而 Dataset 本身仍是唯一权威、可复现的清单。

4.2.3 业务流总览

#

业务流不是一条龙自动跑到底，而是若干段各自独立触发、由 Dagster 编排的流程。核心是把"要等人几天"的标注段单独切出来，别让编排器干等：

flowchart LR
    A["① 上传入湖<br/>定时/文件触发"] --> B["② 预标+标注+质检<br/>派活/收活两段"] --> C["③ 冻结 Dataset<br/>请求触发"] --> D["④ 导出训练格式<br/>随冻结或按需"] --> E["⑤ 训练/评测<br/>人工提交, GPU"]

所有触发都走 Dagster（定时 schedule / GraphQL API / sensor），不再需要 EventBridge。下面几节分别展开关键段，每节配一张小流程图（边上只标动作，细节看正文）。

4.2.4 业务流一：上传入湖（事件触发）

#

上传不能只依赖 OSS 的 ObjectCreated 事件，因为文件出现不代表一次业务上传完整。所以引入 manifest 提交语义。

flowchart TD
    C[客户端] -->|创建 session| GW[上传网关]
    C -->|直传文件| RAW[(OSS Raw)]
    C -->|提交 manifest| GW
    GW -->|写事件| K[(Kafka)]
    RAW -.->|定时/sensor 发现| AF[Dagster]
    AF -->|查状态| TS[(Tablestore)]
    AF -->|提交作业| SP[Spark]
    SP -->|读| RAW
    SP -->|坏数据| Q[(OSS Quarantine)]
    SP -->|写表| ICE[(Iceberg)]
    SP -->|样本本体| LAN[(Lance)]
    SP -->|更新状态| TS

步骤：

1. 客户端向上传网关创建 UploadSession，生成 upload_id，写 Tablestore，并向 Kafka 写 upload_created。
2. 客户端拿预签名 URL，把 MCAP、视频、点云等大文件直传 OSS Raw Zone。
3. 客户端或边端上传 manifest.json（文件列表、大小、sha256、时间范围、robot_id、task_id、schema_version、源头声明的 episode_id，以及声明式 tag 如 scene=kitchen，见 4.2.1.7）。
4. 网关校验 manifest 后向 Kafka 写 manifest_committed，把 session 标记 ready。

触发与执行：定时 T+1（或 Dagster sensor 发现 OSS 新文件）→ Dagster 起 job → 查 Tablestore 状态、提交 Spark → Spark 解析、清洗、切片，写 Iceberg/Lance，更新 Tablestore。这一段把数据加工到 sample 为止。写 Iceberg 基本都是 Spark 在干。

4.2.5 业务流二：标注与质检闭环（人在环）

#

标注是数据平台里唯一"要等人干几天"的环节，也是最容易把编排搞乱的地方。原则是：Dagster 不挂着干等，把流程切成"派活"和"收活"两段短 job，中间几天编排器根本不在场，长期状态存 Tablestore。

flowchart TD
    U[人工/调度] -->|发起标注, 调 Dagster API| AFA[Dagster: job-A 派活]
    AFA -->|GPU 预标| RAY1[Ray auto-label]
    AFA -->|导出脱敏任务包| PKG[(OSS 任务包)]
    AFA -->|建任务+通知| ANNO[Label Studio/CVAT/外包]
    AFA -->|置 LABELING| TS[(Tablestore 批次状态)]
    ANNO -.->|标完数天后提交, webhook 调 Dagster API| AFB[Dagster: job-B 收活]
    AFB -->|导入结果| ICE[(Iceberg: annotation)]
    AFB -->|GPU 自动质检| RAY2[Ray auto-QC]
    AFB -->|抽检 gate| REV[人工审核]
    AFB -->|置 DONE, annotation RELEASED| TS

两段拆解：

• job-A（派活，自动跑完就结束）：Ray 跑 auto-label 生成预标注 → 导出脱敏"标注任务包"（抽帧/降采样，只给需要标的子集，外包不直连数据湖）→ 调 Label Studio/CVAT API 建任务或交付外包 → 通知（钉钉/邮件/webhook）→ 把批次状态置 LABELING → job-A 结束。
• （外包标注，数天）：Dagster 已经睡了，什么都不占用；批次状态 LABELING 存在 Tablestore。
• job-B（收活，被叫醒才跑）：标注方提交时的 webhook 调 Dagster GraphQL API 触发（或由 sensor 轮询批次状态 = RETURNED 兜底）→ 拉结果导入 annotation 表 → Ray 跑 auto-QC → 抽 5% 给人工审核 gate → 通过则 annotation 置 RELEASED、批次置 DONE。

关键点：“人干完了"就是一个调 Dagster API 的外部事件，和"文件到达"“定时到点"同类，不需要引入新的消息总线。 预标模型版本记进 annotation.model_ver，配合 active learning（模型挑难样本优先标）可让人工越来越省。

4.2.6 业务流三：冻结 Dataset 与导出（请求触发）

#

Dataset 不是自动生成的，是按需 freeze，且只挑"标注齐 + 过质检"的样本。

flowchart TD
    U[用户 / 调度] -->|建数据集请求, 调 Dagster API| AF[Dagster]
    AF -->|提交 freeze 作业| SP[Spark]
    SAMPLE[(Iceberg: sample + annotation)] -->|按过滤+质量阈值读| SP
    SP -->|写 dataset + dataset_sample| DS[(Iceberg: dataset)]
    SP -->|置 RELEASED| DS
    AF -->|提交 export 作业| EXP[Spark/Ray 导出]
    EXP -->|物化训练包| OSSE[(OSS: LeRobot/RLDS/...)]
    EXP -->|登记 dataset_export| DS

谁想训练，就发起一次建数据集请求（平台点“建数据集”或 API 写 dataset_requested）→ Dagster 起 job → 一个 Spark 作业按过滤条件（join entity_tag 按 tag 圈数据，通常从一个 saved view 出发，见 4.2.1.7）+ quality_score ≥ 阈值 + annotation RELEASED 挑样本，写 dataset + dataset_sample，算 hash，状态置为 RELEASED。

冻结完紧接着导出：Dataset 只是引用清单，dataloader 喂不进清单，所以随后（或按需）跑一个导出作业，按清单把样本物化成 LeRobot / RLDS / webdataset / HDF5 的 shard 落 OSS，元数据登记进 dataset_export。同一 Dataset 版本可导出多种格式，清单本身仍不可变、可复现。

4.2.7 业务流四：训练与评测（人工 / 调度触发）

#

这是最容易搞错的一段：数据到了不会自动训练，而且训练不在 Spark 上跑。

flowchart TD
    U[用户 / 调度] -->|提交 TrainRun + 版本, 调 Dagster API| AF[Dagster]
    AF -->|提交 GPU 作业到 K8s+Volcano| GPU[Ray Train / GPU 作业<br/>非 Spark]
    DSX[(OSS: dataset_export 训练包)] -->|直接喂| GPU
    LAN[(Lance)] -.->|或直读样本本体| GPU
    GPU -->|产物| ART[ModelArtifact]

• 训练不是数据到了就自动跑，而是由人 / CI / 调度显式提交一个 TrainRun，指定 dataset_name + version。
• 训练作业本身不是 Spark。Spark 擅长批量 ETL；模型训练（尤其深度学习）跑在 GPU 训练框架（PyTorch 等）+ GPU 集群（K8s / PAI）上，是独立作业。数据平台的职责到“产出一个 RELEASED 的 Dataset”为止，训练只是消费它。
• 评测 EvalRun 类似：推理在 GPU 作业上跑，指标聚合和报告可以再用 Spark / Trino SQL 出。

各组件分工因此很清楚：

• Dagster：统一触发与编排（定时 schedule、sensor、人工请求走 GraphQL API），管理 asset 版本与血缘，不做计算。
• Spark：数据 ETL、质量检查、切片、Dataset freeze、指标聚合——写 Iceberg 的活几乎都是它。
• Ray：GPU/ML 作业——自动预标、自动质检、后处理、训练、评测。
• GPU 调度（K8s + Volcano）：gang scheduling、配额、优先级、抢占。
• Iceberg / Lance / OSS / Tablestore / Kafka：存储与状态，职责同前。

最关键的一点：Dataset 必须是不可变发布物。 训练和评测引用的是 dataset_name + dataset_version + manifest_hash，而不是“某个目录下当前所有文件”。这样模型效果变化时，才能追溯到底是数据、代码、参数还是标注变了——这正是整个数据平台存在的意义。

4.2.8 数据一致性：以 Iceberg commit 为准

#

一次处理要同时写好几个地方：OSS 的数据文件、Lance 的样本文件、Iceberg 的元数据、Tablestore 的状态、Kafka 的事件。这是个典型的多写一致性问题，如果不定规则，Spark 写完 OSS 就崩溃时，就会留下“OSS 有文件但没人引用”的孤儿文件，或者“Iceberg 有指针但文件不全”的悬空引用。

这里的原则是：不要追求跨存储的分布式事务，而是指定一个唯一的“真相源”，其余都向它对齐。 具体做法：

• 真相源是 Iceberg commit。一份数据“算不算存在、能不能用”，以它有没有被 Iceberg 快照引用为准，OSS/Lance 里有没有文件都不算数。
• 写入顺序固定：先写文件，最后提交元数据。Spark 先把数据文件写进 OSS/Lance（用内容 hash 或 uuid 命名，保证可重复、不覆盖），全部写完后再做一次 Iceberg 原子 commit（Iceberg 本身是 ACID 的）。commit 成功，这批数据才“出现”；commit 失败，前面写的文件就是没人引用的孤儿，等着被清理。
• 孤儿文件靠 GC 兜底。定期跑 Iceberg 的 expire_snapshots 和 remove_orphan_files，把没有被任何快照引用的文件清掉。所以中途失败不会污染数据，只会浪费一点临时空间。
• 作业幂等，按目标分区覆盖重写。每个作业按 upload_id 或分区幂等：重跑时覆盖写同一批目标（Iceberg overwrite/MERGE），而不是追加，这样“至少一次 + 幂等 = 最终一致”，不必强求 exactly-once。
• Tablestore / Kafka 是派生态，可重建。Tablestore 状态即使丢了，也能从 Iceberg 当前快照 + Kafka 事件重放重建；Kafka 事件是 at-least-once，消费端按业务 id 去重。它们都不是真相源，所以它们和 Iceberg 短暂不一致是允许的，最终会对齐。

一句话：文件先落、Iceberg 最后原子提交、未提交即不存在、孤儿定期回收、重跑幂等覆盖。 一致性问题就被收敛成“以 Iceberg 快照为准”这一条规则。

4.2.9 当前架构待补强的点

#

这一版架构骨架是成立的，但下面这些点是后续必须补的设计取舍，否则规模上来会出问题：

• Episode 边界 ≠ 上传边界：一次采集可能跨多文件、多次上传，一次上传也可能含多个 Episode。靠 episode_file 关联表 + 按 robot_id/时间窗/task_id 聚合来解耦，不要假设“一次上传就是一个 Episode”。
• 分层物化成本：Raw+Bronze+Silver+Gold+Dataset 若每层都全量物化，1PB 会膨胀到 3–5PB。要明确哪些层物化、哪些只是 Iceberg 视图或轻量派生（判据见 4.2.1.3：高频/复杂 join 的物化，探索性低频的用视图现拼）；Raw 配生命周期降冷归档，Bronze 可重建因此可设较短保留期。
• 不可变 vs 删除/合规：Raw 只追加，但现实有“标定错了要删”“隐私数据要下线”。需要补失效机制：分区级删除或 tombstone 标记 + 受影响 Dataset 重新 freeze，并保留删除审计。
• 血缘怎么落地：就用 dataset_sample → sample → episode → episode_file → raw_file 这条链，支持从模型反查到具体原始文件，不要只在文档里说“有血缘”。
• schema 演进：MCAP 的 topic/message schema 会变。Bronze 用 Iceberg 的 schema evolution（加列、宽化类型）+ schema_version 列做兼容，避免改一次结构就重刷历史。
• 数据可用性契约：给下游一个明确信号，而不是让训练方猜。例如 Dataset 状态机 BUILDING → RELEASED → DEPRECATED，只有 RELEASED 且 quality_score ≥ 阈值 才允许被训练/评测引用。
• Quarantine 闭环：坏数据进隔离区后要有出口——记录隔离原因、人工或规则修复、重新走 Bronze、超期告警，别让它变成只进不出的黑洞。
• 标注质量与成本：多人/外包标注要有一致性校验（多标抽检、Krippendorff’s α 之类）和 auto-label 预标降本；标注也要版本化（anno_version），重标是新增而不是覆盖历史。
• GPU 调度与配额：训练、自动预标、自动质检抢同一批卡，要用 Volcano/Kueue 做队列、配额、优先级与抢占（训练长作业 gang scheduling、推理短作业可被抢占），否则大作业会饿死小作业。
• 导出格式维护：LeRobot/RLDS/webdataset 等格式会随训练框架演进，dataset_export 要记录 format 版本，换格式是新增导出而不是覆盖旧包。

4.2.10 状态与监控

#

数据平台自己的 Tablestore 只存数据业务状态，例如 UploadSession、IngestJob、AnnotationTask（标注批次）、DatasetBuild、EvalRun、TrainRun、人工修正记录，它是可以从 Iceberg + Kafka 重建的辅助状态，不是权威数据。

初期监控也要保留基础项：

• 上传成功率、manifest 校验失败率、坏文件数量。
• Kafka lag、Dagster run 失败率、Spark 作业耗时、重试次数。
• Raw 到 Dataset 的端到端延迟，T+1 是否按时完成。
• Iceberg 小文件数量、compaction 成功率、查询延迟。
• OSS 存储量、生命周期降冷比例、单日新增成本。
• Dataset 发布成功率、样本数量、质量评分分布、评测/训练失败率。
• 标注：任务周转时长、外包交付率、标注返修率、auto-label 采纳率、抽检一致性。
• GPU：卡利用率、训练/预标作业排队时长、抢占次数、失败重试率。
• 导出：dataset_export 成功率、训练包体积、导出耗时。

4.3 平台工程横切面

#

前面讲的是数据怎么流；这一节讲让平台"规模上来不失控"的横切面设计：可追踪、桶规划、查询、安全、配置驱动、状态存储、故障感知。

4.3.1 可追踪性：批次号与版本五元组

#

任何一次工作流和处理都必须可追踪。 每次运行固化五元组：

落地方式，写进数据而不是只写日志：

• 每张表加审计列：_batch_id, _run_id, _ingested_at, _source_uri——任何一行数据都能反查到是哪个批次、哪次运行、哪个源文件产生的。
• Iceberg commit 属性：提交快照时把 batch_id/run_id 写进 snapshot summary，表级历史也可追。
• 编排器自带一部分：Dagster asset 天然带 code_version/data_version，每次物化都是一条带版本的事件记录（这是选它的理由之一，见 4.1.4）。

4.3.2 桶与路径规划，以及 Iceberg 的边界

#

命名规范（环境必须在桶级隔离，权限才好做；地区/业务方按需加维度）：

oss://{公司}-{env}-lake-{layer}[-{region}]/{业务域}/{实体}/...
例：acme-prod-lake-silver-cnhz/teleop/episode/...

Raw 区非表数据（MCAP/视频）路径再按 {业务}/{批次}/{实体} 组织，例如 raw/teleop/2026-07-03/upload-123/r-001/xxx.mcap。

Iceberg 的边界——它管了什么、什么仍是你的设计责任：

一句话：Iceberg 表 location 以下不用操心，location 以上（桶/域/实体映射）和 Raw 区路径是你的设计责任。

4.3.3 TB/PB 级查询：永远不扫全量，三种查询路由

#

PB 级湖上"能做什么” = 批处理全量（Spark，小时级）+ 裁剪后的交互查询；“不能做什么” = 亚秒点查和无分区条件的全表扫。按查询形态路由：

裁剪生效靠日常运维：compaction 到 128–512MB 目标文件、按常用过滤列 sort/z-order、分区键选 date + robot_id 这类高频过滤维度。TB 级交互查询没问题，PB 级只要裁剪后实际扫描量是 GB~TB 就行。

4.3.4 安全与授权：入口收敛，一处管权限

#

• 收敛原则：人类用户只能通过查询引擎 + Catalog（Trino + DLF，需要行列级再加 Ranger/OPA）访问数据，禁止直接读桶；桶级 RAM policy 只授给平台的 service role。授权只需在一个地方做。
• 粒度：DLF/Ranger 做库/表/列级；行级过滤和列脱敏用引擎能力（Trino row filter / column mask）或视图。
• 脱敏配合分层：Silver 默认产出脱敏版；敏感原文单独成表（或敏感列独立），只授极少数角色。“一些人看得到、另一些人看不到"就变成角色 → 表/列的映射问题。
• 审计：查询日志 + ActionTrail，谁查了什么可追。

4.3.5 热配置与静态配置：配置驱动架构

#

配置驱动的本质是"数据化”——把硬编码在脚本里的业务逻辑（字段映射、同步策略、脱敏规则）提取为数据，让代码变成通用引擎、配置变成业务声明。加新数据源不改代码，只加配置——开闭原则在数据工程的体现。划分判据：改动会改变计算逻辑的控制流 → 代码（静态）；改动只是参数/映射/规则 → 配置（热）。

两个防坑：

• 配置本身也要版本化和审计：config_ver 是 4.3.1 五元组之一，某批数据出问题能查到当时用的哪版映射。
• 别让配置图灵完备：一旦配置里要写 if/else 表达式，说明这段逻辑该回到代码，否则只是发明了一门更烂的编程语言。

Dagster 的 Components/YAML 声明式恰好是这个模式的官方落点：加一个新数据源 = 加一个 YAML（见 4.1.4）。

4.3.6 状态与审计存储：三类状态，三个去处

#

一个常见疑问：“Iceberg 里存的是数仓表，那 DDL 审计、操作日志这类东西放哪？” 答案是——审计日志本质就是追加型表数据，它就该是 Iceberg 表，只是放独立的 ops namespace（库），和业务数据分开管权限。“数仓表"这个直觉太窄了：Iceberg 存的是"所有值得 SQL 分析的表”，运维表也是表。但确实不是所有日志都进湖，按用途分三类：

模式是双写：最新态进 Tablestore（排障按 batch_id 一查便知），历史沉淀进 Iceberg（“过去 30 天哪个源失败率最高"用 SQL 答）。

DDL 的"真相"其实有三层，各司其职、不冲突：

配套规矩：所有 schema 变更必须走 git 里的 migration 脚本（禁止控制台手改），执行时写一行 ddl_changelog。

4.3.7 故障感知：分层错误处理、DLQ 与三类告警

#

出了故障必须有人感知到，而不是死在那里。原则：坏一条不坏一批，坏一批不坏管道。

最容易漏的是"该跑没跑”——没跑的任务不会触发失败告警。所以告警必须覆盖三问：

1. 挂了吗：失败回调（任务级）。
2. 没跑吗：deadman/新鲜度检查——“T+1 数据早上 9 点还没 RELEASED 就告警”（Dagster freshness check 原生支持，OSS 版可用）。
3. 跑了但不对吗：质量规则（行数骤降、空值率、分布漂移）作为管道内的 check 步骤，不过阈值就挡住下游并告警。

4.3.8 SoT 的域划分：Dagster asset 和 Iceberg 冲突吗？

#

引入 Dagster 后有一个必须回答的问题：Dagster 的 asset 概念会不会和"Iceberg 是真相源”（4.2.8）冲突？多套元数据（Dagster 元数据库、DLF Catalog、Iceberg metadata、Tablestore、配置表）怎么统一管理？

先说结论：不冲突，因为它们回答的是不同域的问题。 Dagster 的 asset 不是数据，是"对一张 Iceberg 表的编排视角描述"——它记录的是这张表怎么生产、依赖谁、上次什么时候用哪版代码物化的。这些是生产过程的事实，不是数据本体的事实。数据本体的真相仍然只有一个：Iceberg commit，4.2.8 的原则一个字不用改。

SoT 的统一管理，解法不是"只留一个存储"，而是按域划分：每个域声明唯一 SoT，其余全部是可重建的派生投影：

检验方法一句话：问"这个存储丢了会怎样"。 Dagster 元数据库丢了——数据一行不少、正确性零影响，损失的只是物化历史和血缘展示，重跑可重建。这恰恰证明它不是数据 SoT，天然不构成冲突。

冲突只会在一种情况下出现：让 Dagster 的元数据回答数据域的问题——比如用"Dagster 有没有物化记录"判断"这批数据存不存在"。这是错的，数据存在与否永远查 Iceberg 快照。规则定死：谁的域谁说了算，跨域只读不裁决。

实操上再加一个把两个世界钉在一起的连接件：asset key 和 Iceberg 表名一一对应，每次物化时把 Iceberg 的 snapshot_id 写进 Dagster 的 materialization metadata（与 4.3.1 五元组的 input_snapshot_id 呼应）。这样从 Dagster UI 看到一次物化，能精确跳到对应的 Iceberg 快照；从一个 Iceberg 快照，也能反查是哪次 run 产生的。两套元数据不是竞争关系，是互相引用的关系。

结尾

#

唉，尴尬

狗头的赞赏码.jpg