具身智能数据工厂实战：基于阿里云OSS、MaxCompute、DataWorks与PAI的会生长架构

具身智能这个词这两年从论文里一路烧到了工程落地现场，做机器人本体的、做灵巧手的、做具身大模型的团队都在抢同一个东西——数据。不是那种网上随便爬爬的图文对，而是带关节角度、力矩反馈、多视角RGB-D、触觉阵列、时间戳对齐的"真机轨迹数据"。这类数据的采集成本高得离谱，一条高质量的抓取轨迹可能要反复遥操作几十次才能拿到，而训练一个能泛化的策略动辄需要几十万甚至上百万条。问题就卡在这：数据从哪来、怎么存、怎么洗、怎么喂给训练任务，还要保证整条链路能随着数据量增长而"自己长大"。

"息壤开物 × 阿里云"这个项目要解决的就是这件事——给具身智能造一座"会生长的数据工厂"。息壤开物负责机器人侧的数据采集、标注和仿真合成，阿里云这边用OSS做海量非结构化数据的底座，MaxCompute扛批量清洗和特征工程，DataWorks编排整条数据流水线，PAI承接模型训练和仿真策略迭代。整套东西不是一次性搭完就完事，而是随着数据规模、任务类型、团队人数变化持续演进。下面我按实际搭建和踩坑的顺序，把这座工厂怎么从零长起来讲清楚。

1. 具身数据工厂到底难在哪：先看清问题的形状

1.1 具身数据和普通AI数据不是一回事

很多人第一反应是"不就是存数据吗，OSS一挂不就完了"。真上手就会发现具身数据和CV、NLP那套完全不是一个物种。一条完整的机器人操作轨迹，包含的模态至少有：关节位置/速度/电流（通常7自由度机械臂就是7维，加上夹爪就是8维）、末端执行器六维力/力矩、腕部相机和外部相机的RGB视频流、深度图、有时还有触觉传感器的压力分布矩阵。这些数据在时间轴上必须严格对齐，采样率还各不相同——关节控制环可能1000Hz，相机30Hz，触觉可能100Hz。

这就带来第一个硬骨头：时间对齐和插值。你不能简单地把所有流按最小时间戳切齐，那样会丢信息；也不能各存各的，训练时再对齐，那样IO开销爆炸。实际做法是在采集端就做一次粗对齐，落到OSS时按"episode"为单位组织，每个episode里各模态分目录存，附带一份manifest文件记录各流的采样率和起始时间戳。manifest用JSON存，训练时先读manifest再决定怎么重采样。

第二个硬骨头是数据量级。一个episode如果录30秒，双相机RGB加深度，压缩后大概200MB到500MB。要凑100万条episode，就是200TB到500TB的原始数据。这还没算仿真合成数据，仿真数据量可以轻松翻十倍。所以存储层必须是对象存储，不能是块存储或者文件系统，成本差一个数量级。

1.2 "会生长"这三个字才是真正的设计约束

大部分数据平台是"建好就用"，但具身智能这个领域变化太快。今天采的是桌面抓取，明天可能要采双臂协作，后天要加移动底盘。数据schema会变，模态会加，标注规范会改。如果一开始把表结构写死，三个月后就得推倒重来。

"会生长"落到工程上，意味着三件事：schema要能演进、流水线要能插拔、算力要能弹性伸缩。schema演进靠MaxCompute的分区表加版本化字段设计；流水线插拔靠DataWorks的节点编排和依赖管理；弹性伸缩靠PAI的按需资源和OSS的无上限容量。这三件事后面会分别展开。

提示：别一上来就追求"完美schema"。具身数据的字段几乎必然会变，先按最小可用集建表，留好扩展字段（比如一个JSON类型的extra列），比反复改表结构划算得多。

1.3 谁适合参考这套方案

这套东西不是给个人玩家准备的，个人玩机器人用本地硬盘加几个脚本就够了。它适合的是：有真机采集设备（至少一台机械臂加相机）、团队规模5人以上、数据量已经超过10TB、并且有持续训练策略需求的团队。如果你还在单机阶段，先别上云，把采集流程和标注规范跑顺了再说。上云是为了解决规模和协作问题，不是为了赶时髦。

2. 存储底座：OSS怎么组织才不会被自己坑死

2.1 目录结构设计：episode为原子单位

OSS是扁平的key-value存储，所谓"目录"只是key的前缀。但前缀设计直接决定了后续MaxCompute和PAI读取数据的效率。我试过几种方案，最后稳定下来的是按"任务类型/采集批次/episode_id/模态"四级组织：

oss://embodied-data/ task=grasp/ batch=20240115_robot01/ episode=000001/ manifest.json joint_states.parquet camera_wrist.mp4 camera_ext.mp4 depth_wrist.zarr tactile.parquet episode=000002/ ...

用task=和batch=这种带等号的前缀，是为了后续MaxCompute建外部表时可以直接用分区字段解析，不用额外写解析逻辑。episode_id用零填充的六位数字，保证字典序和时间序一致，列目录的时候不会乱。

manifest.json里记录这个episode的元信息：采集时间、机器人型号、任务描述、各模态的采样率、时长、标注状态、质量评分。这个文件很小但极其重要，它是数据治理的抓手。后面做数据筛选、质量过滤、训练集划分，全靠读manifest。

2.2 存储类型和生命周期：别让冷数据吃掉预算

OSS有好几种存储类型：标准、低频、归档、冷归档。具身数据的特点是"采的时候热，训的时候热，中间可能躺几个月"。刚采完要马上做质量检查和标注，这是热；标注完可能等某个训练任务才用，这是温；训练完的原始数据可能半年都不碰，这是冷。

我的做法是配生命周期规则：采集后30天内保持标准存储，30到90天转低频，90天以上转归档。归档存储的取回有延迟（分钟级到小时级），所以训练任务启动前要提前触发取回。这里有个坑：归档存储的取回是按量计费的，如果训练任务频繁随机读取归档数据，取回费用可能比存储费还高。所以策略是：训练集确定后，把要用的数据批量取回到标准存储，训练期间不再动归档层。

2.3 上传链路：断点续传和并发控制

真机采集现场的网络往往不稳定，采集端可能是工控机或者边缘盒子。直接调OSS SDK上传大文件，一旦断网就得重传，几百MB的episode重传几次人就疯了。必须用分片上传加断点续传。OSS的SDK支持multipart upload，把大文件切成比如8MB的分片，记录已上传的分片ID，断网后从断点继续。

并发数也要控制。采集端如果同时上传多个episode，把带宽打满会影响采集本身的稳定性（相机掉帧、控制延迟）。我的经验是采集端上传并发限制在2到3个，并且给上传进程设一个带宽上限，留出余量给采集。这个细节看起来小，但实际现场因为上传把采集搞崩的情况我见过不止一次。

注意：分片上传产生的碎片如果不清理会一直占存储。OSS有生命周期规则可以自动清理未完成的分片上传，一定要配上，否则几个月后你会发现账单里有一堆看不见的碎片。

3. MaxCompute：把TB级原始数据洗成能训练的样子

3.1 外部表挂OSS：先能查，再谈洗

数据在OSS里是文件，MaxCompute要处理它，第一步是建外部表。MaxCompute支持OSS外部表，可以直接把OSS上的Parquet、CSV、JSON映射成表来查。具身数据里结构化程度最高的是关节状态和力/力矩，这些我存成Parquet，直接建外部表就能用SQL查。

建外部表的关键是分区。前面OSS目录里的task=和batch=就是分区字段，建表时声明成分区列，查询时加分区过滤，能大幅减少扫描量。比如要查某个批次的所有episode，where batch='20240115_robot01'，MaxCompute就只扫这个前缀下的文件。

CREATE EXTERNAL TABLE IF NOT EXISTS ods_episode_joint ( episode_id STRING, timestamps ARRAY<BIGINT>, joint_positions ARRAY<DOUBLE>, joint_velocities ARRAY<DOUBLE>, joint_currents ARRAY<DOUBLE> ) PARTITIONED BY (task STRING, batch STRING) STORED AS PARQUET LOCATION 'oss://embodied-data/';

这里有个细节：关节状态我用ARRAY类型存，因为一个episode里是变长的时间序列。MaxCompute对ARRAY的支持够用，但如果要做复杂的时序窗口计算，ARRAY不如展开成行来得方便。所以实际流程是：外部表先做粗筛（按分区、按manifest里的质量分），筛出来的数据再用一个转换节点展开成"一行一个时间步"的宽表，后续特征工程都在宽表上做。

3.2 数据清洗：具身数据特有的脏法

具身数据的脏和互联网数据完全不一样。互联网数据脏在缺失、重复、格式乱；具身数据脏在物理上不合理。常见的几类：

• 关节角度跳变：相邻时间步角度差超过物理极限，通常是编码器丢数或者通信丢包。
• 力/力矩尖峰：力传感器在碰撞或者标定漂移时会出现远超量程的读数。
• 时间戳回退：多传感器时钟不同步，偶尔出现时间戳倒退。
• 相机丢帧：视频流中间缺帧，导致和关节数据对不齐。

这些用SQL都能筛。比如关节跳变，展开成宽表后算相邻行的角度差，超过阈值就标记这个episode为低质量。力尖峰同理，超过量程95%的读数计数，超过一定比例就剔除。时间戳回退直接检测单调性。

-- 标记关节跳变严重的episode SELECT episode_id, SUM(CASE WHEN ABS(joint_pos - LAG(joint_pos) OVER (PARTITION BY episode_id ORDER BY ts)) > 0.5 THEN 1 ELSE 0 END) AS jump_count FROM dwd_joint_wide GROUP BY episode_id HAVING jump_count > 10;

清洗不是一次性的，是持续跑的。新数据进来就触发清洗，清洗结果写回一张质量表，manifest里的质量评分也从这里更新。这样数据湖里始终有一份"干净数据视图"，训练任务只读干净数据。

3.3 特征工程：把原始信号变成策略能吃的输入

策略网络不直接吃原始关节电流，它需要的是有物理意义的特征。常见的特征工程包括：末端执行器的笛卡尔位姿（从关节角度正运动学算出来）、末端速度、抓取力的大小和方向、相机图像的降采样特征（这个通常不在MaxCompute做，太重了，放到PAI）。

MaxCompute适合做的是数值型特征的批量计算。比如正运动学，给定DH参数和关节角度，用SQL的UDF或者Python UDF算末端位姿。UDF用Python写，MaxCompute支持Python UDF，把机器人运动学库打包进去就行。算出来的特征写回一张特征表，训练时直接读。

这里有个性能经验：特征计算尽量向量化，别一行一行算。MaxCompute的SQL引擎对批量操作优化得好，但Python UDF是逐行调用的，如果UDF里再套循环，几百万行能跑到天荒地老。我的做法是把一个episode的所有时间步打包成一个ARRAY传给UDF，在UDF内部用numpy向量化算完再返回，这样调用次数从百万级降到万级。

4. DataWorks：让整条流水线自己跑起来

4.1 编排逻辑：从"数据到达"到"可训练"的完整链路

DataWorks在这套体系里扮演的是"调度大脑"。整条链路大概是：OSS有新episode上传 → 触发清洗任务 → 清洗完更新质量表 → 质量达标的进入特征工程 → 特征写完更新训练集索引 → 通知PAI可以拉数据训练。

这个链路用DataWorks的依赖关系串起来。OSS上传完成可以通过事件触发，也可以用DataWorks的定时调度加一个"检查新文件"的节点。我倾向于后者，因为事件触发在跨服务时偶尔会丢，定时检查虽然有一点点延迟（比如5分钟一次），但稳定可控。

每个节点都是一个MaxCompute SQL任务或者Python任务，节点之间用依赖连线。DataWorks的好处是它自带重跑、补数据、监控告警。某个节点挂了，能自动重试，重试还不行就告警到群里。补数据功能在schema变更后重刷历史数据时特别有用。

4.2 增量 vs 全量：别每次都重算

数据工厂最容易犯的错是"每次全量重算"。数据量小的时候没感觉，到了几百TB，全量重算一次要几个小时甚至一天，完全没法用。必须做增量。

增量的关键是分区裁剪加水位线。每次调度只处理"上次处理之后新增的分区"。DataWorks里可以用一个控制表记录每个任务的处理水位，任务开始时读水位，处理完更新水位。MaxCompute的分区表天然支持按分区读，只读新分区，扫描量就下来了。

但增量有个坑：特征工程有时需要跨episode的上下文。比如归一化要用全局统计量（均值、方差），这个统计量如果只用增量数据算，会漂移。解决办法是定期（比如每天）全量重算一次统计量，增量任务用这个统计量做归一化。统计量本身很小，全量重算成本可接受。

4.3 数据质量监控：让问题在训练前暴露

具身数据最怕的是"训练跑完了才发现数据有问题"。所以DataWorks里要挂质量监控节点，在数据进入训练集之前做检查。检查项包括：episode数量是否达标、各模态时长是否一致、质量评分分布是否正常、特征是否有NaN或无穷值。

这些检查用SQL写规则，DataWorks有数据质量模块可以配置。比如"特征表中NaN比例超过1%就告警"，"某个批次episode数量比上周下降超过30%就告警"。告警直接推到团队群里，别等到训练任务失败才发现。

提示：质量监控的阈值别设太死。具身数据本身波动大，阈值太严会天天告警，最后大家就麻木了。我的做法是分两级：警告级（记录但不阻断）和阻断级（直接停流水线）。只有会导致训练崩溃的问题才设阻断级。

5. PAI：训练和仿真迭代怎么接上数据工厂

5.1 数据读取：从OSS直读还是走缓存

PAI训练任务读数据有几种方式：直接从OSS读、通过MaxCompute读、或者用PAI的数据集缓存。具身数据的训练通常是"读一个episode的多个模态，做数据增强，喂给策略网络"。这种读取模式是随机访问，不是顺序扫描，直接从OSS读延迟高。

我的做法是：训练前把本次要用的episode批量取回到一个高速缓存层（可以是OSS的低频转标准，或者PAI的本地缓存），训练时从缓存读。PAI支持挂载OSS到训练容器，也支持数据集加速。对于小规模训练（几万条episode），直接挂OSS够用；对于大规模（几十万条以上），一定要用缓存，否则GPU利用率会被IO拖垮。

5.2 仿真合成数据：数据工厂的"产能放大器"

真机数据贵，仿真数据便宜。具身智能里仿真合成是数据量的主要来源。息壤开物这边用仿真环境批量生成轨迹，生成的格式和真机数据保持一致，落到同一个OSS目录结构里，只是manifest里标记source=sim。

仿真数据的问题是域差异（sim-to-real gap）。仿真里的物理参数和真实世界有偏差，直接混着训练效果不好。常见做法是域随机化：仿真时随机化摩擦系数、质量、光照、相机噪声等参数，生成大量多样化的数据。这些随机化参数也记在manifest里，训练时可以做域自适应。

PAI这边承接的是策略训练和仿真策略的迭代。训练任务从数据工厂拉数据，训练完的模型评估结果写回一张表，DataWorks监控这张表，如果新模型比旧模型好，就触发下一轮数据采集（针对模型表现差的场景多采数据）。这就形成了"采集-训练-评估-再采集"的闭环，也就是"会生长"的真正含义。

5.3 成本控制：训练任务是最烧钱的环节

PAI的GPU资源按量计费，一个训练任务跑几十张卡几个小时，费用很可观。控制成本有几个抓手：一是数据预处理尽量在CPU侧做完，别让GPU等数据；二是用竞价实例跑容错性好的训练任务，成本能降不少；三是训练任务设超时和早停，别让一个跑飞的实验烧一整晚。

我踩过的一个坑是：训练任务读数据时没做shuffle，所有worker读同一批数据，导致GPU利用率忽高忽低。后来在数据加载层加了全局shuffle，并且按worker数分片，利用率才稳定下来。这个细节在数据量小的时候看不出来，数据量一大就是灾难。

6. 让工厂"生长"的几个关键设计

6.1 schema演进：加字段不炸历史数据

具身数据加模态是常态。今天加触觉，明天加音频。如果表结构是强schema，加字段就要改表、重刷历史数据，成本极高。我的做法是：核心字段用强schema，扩展字段用JSON。比如关节状态、时间戳这些必须有，用列存；触觉、音频这些后加的，先塞进一个extra的JSON列，等稳定了再提升为独立列。

MaxCompute对JSON有支持，可以解析JSON里的字段来查。这样加模态时不用改表结构，历史数据的extra为空也不影响。等某个模态稳定了，再写一个转换任务把它从JSON里抽出来变成独立列，历史数据补上默认值。

6.2 流水线插拔：新任务类型怎么接进来

新任务类型（比如从抓取扩展到装配）进来时，不应该改动已有流水线。DataWorks的节点是独立的，新任务类型就是新加一组节点，复用已有的清洗和特征工程模板，只是参数不同。模板化是关键：把清洗逻辑、特征计算逻辑写成参数化的SQL模板或者Python脚本，新任务类型传入不同的参数就能跑。

这样扩展的代价就是"配置一个新任务"，而不是"开发一套新流程"。团队里新人也能快速上手，不用理解全部细节。

6.3 算力弹性：闲时省钱，忙时扩容

数据工厂的负载是波动的。采集高峰期上传量大，清洗任务排队；训练高峰期GPU吃紧；平时可能都很闲。MaxCompute和PAI都支持按量付费，闲时不用不花钱。但要注意预留资源 vs 按量资源的权衡：如果每天都有稳定的清洗任务，预留一些CU（计算单元）比全按量便宜；如果负载波动大，全按量更灵活。

我的经验是：清洗和特征工程这类每天必跑的，预留基础CU；训练这类突发性的，用按量加竞价。OSS没有预留概念，按量就行，配合生命周期规则控制成本。

7. 实操中踩过的坑和几条硬经验

7.1 时间戳对齐别在云端做

一开始我想着"数据都上云了，对齐也在云上做吧"，结果发现云端做对齐要反复读多个模态的文件，IO开销巨大，而且对齐逻辑复杂，SQL写起来很痛苦。后来改成采集端做粗对齐，云端只做校验。采集端在录数据时就用统一时钟打时间戳，各模态按同一时钟记录，落盘时已经大致对齐。云端只检查对齐质量，不合格的打回重采。这个改动让云端处理量降了一大半。

7.2 manifest是数据治理的生命线

manifest.json这个文件看起来不起眼，但它是整个数据工厂的索引。质量评分、标注状态、来源（真机/仿真）、任务类型、采集设备，全在里面。训练集划分、数据筛选、问题追溯，都靠读manifest。所以manifest的schema要严格管理，写入时要校验，不能随便加字段。我建议给manifest定义一个JSON Schema，写入前校验，避免脏manifest污染整个数据集。

7.3 别忽视小文件问题

具身数据如果按时间步存，会产生海量小文件。OSS对小文件不友好（每个文件都有元数据开销），MaxCompute读大量小文件也慢。解决办法是按episode聚合存储，一个episode的关节数据存成一个Parquet文件，而不是一个时间步一个文件。如果episode内部数据量还是太小，就按批次合并。我一般控制单个Parquet文件在64MB到256MB之间，这个区间读写效率最好。

7.4 权限和协作：别让数据成为孤岛

团队大了之后，数据权限是个大问题。采集团队、标注团队、训练团队、仿真团队，各自需要不同的访问权限。OSS有RAM策略，MaxCompute有项目空间权限，DataWorks有节点权限。要提前规划好角色：采集团队只写不读（防止误删），标注团队读写标注目录，训练团队只读训练集目录。权限没规划好，要么数据泄露，要么互相干扰。

7.5 监控要覆盖全链路

最后一条，监控别只盯训练任务。从采集端的上传成功率、OSS的存储增长、MaxCompute的任务耗时、DataWorks的调度延迟，到PAI的GPU利用率，全链路都要有监控。任何一环出问题，都会影响最终的数据产出。我习惯在DataWorks里挂一个"全链路健康检查"节点，每天跑一次，把各环节的关键指标汇总成一张表，异常就告警。这样不用天天盯着各个控制台，有问题主动找上门。

这套东西搭下来，从最初的手忙脚乱到现在的相对稳定，大概花了小半年。最大的体会是：具身数据工厂的核心不是某个云产品，而是数据流的组织方式。OSS、MaxCompute、DataWorks、PAI都只是工具，真正决定成败的是你有没有想清楚数据从采集到训练要经过哪些环节、每个环节的输入输出是什么、schema怎么演进、质量怎么保证。想清楚这些，工具选型反而是水到渠成的事。