本文以视觉清洗管线为例，介绍如何使用百舸工作流编排一条多步骤数据管线：从预置模板创建工作流，配置队列和存储参数，运行并查看各节点的状态、日志与监控，最后查看输出结果。平台公共 BOS 上自带样例数据，您无需准备任何数据即可跑通全流程。文末介绍如何将样例数据替换为您自己的数据。

背景信息

什么是百舸工作流

百舸新一代工作流是百舸平台上用于编排端到端数据管线的模块，开箱即用，帮助算法工程师高效地组织、管理和执行复杂的机器学习工作流，覆盖从数据准备、模型开发到模型部署的全生命周期。

工作流将数据处理各环节中分散的步骤组织为一条有向无环图（DAG）流水线：每个节点是一个独立的任务，节点之间的连线定义执行顺序和依赖关系。工作流定义一次即可反复运行，也支持定时运行和失败后从断点续跑。

为什么使用工作流

在具身智能数据生产中，一条管线通常包含解码切分、规则过滤、模型推理、格式导出等十几个步骤，CPU 步骤和 GPU 步骤交替出现，数据量动辄几十 TB。如果将所有步骤放在单个作业中运行，会遇到以下三个问题，工作流分别提供了对应的能力：

• 资源按节点分配：单个作业中所有步骤共用同一批资源，CPU 步骤运行时 GPU 处于闲置。工作流中每个节点单独配置资源，只有需要 GPU 的节点才申请 GPU。
• 失败后从断点续跑：单个作业中一步失败，往往需要从头重跑。工作流可以从失败的节点继续运行。
• 按节点排查问题：单个作业中步骤边界不清，排查困难。工作流中每个节点都有独立的运行状态、日志、监控和中间文件。

本文示例：视觉清洗管线

本文使用视觉清洗管线作为示例。视觉清洗管线是百舸工作流具身系列数据管线的入口，对应预置工作流模板 人类演示数据视觉预处理与质量清洗。

在视频进入手部重建、VLM 打标等成本较高的 GPU 环节之前，该管线先用 CPU 剔除模糊、静止、过短的片段。输出的 clips.jsonl 可直接作为下游管线 「人类演示视频转 LeRobot v3.0」和 「多模态 Caption 标注」的输入。

前提条件

在控制台操作之前，请逐项确认以下准备工作。

• 已开通百舸工作流服务。
• 已创建全托管资源池和资源队列，详情请参见创建全托管资源池。
• 已创建 PFS 实例并绑定至上述资源池，用于存放中间结果和最终输出，详情请参见PFS 存储配置。
• （可选）如需查看结果文件，需准备一台挂载同一个 PFS 实例的开发机，详情请参见开发机存储挂载与数据集挂载。
• （可选）如需使用自己的 BOS 数据，请参见下文“使用自有数据运行管线”中对应方式的准备事项。

工作原理

运行时，平台按 DAG 顺序将每个节点（在控制台中称为“子任务”）提交为一个独立的分布式任务。在视觉清洗管线中，每个节点是一个 RayJob：由 1 个 Head 和 N 个 Worker 组成一个临时 Ray 集群，节点结束后集群即释放，下游节点再拉起自己的集群。

节点之间不直接传递数据，而是通过共享存储上的文件交接：上游节点将结果写入约定路径，下游节点从该路径读取。

核心处理流程

视觉清洗管线包含 6 个节点，各节点的功能、默认参数和输出文件如下表所示。

说明

• step1–step5 的输出文件写在 tmp_dir 下；step6 的最终产物写入 output_file 指定的路径。
• 全局参数 no_undistort=true 时，step1 直接跳过。
• 每个节点读取上一节点写出的 JSONL，在原有列的基础上追加新列后写出，因此任意一个中间文件都可以单独打开排查。

操作流程

步骤一：从模板创建工作流

1. 登录百舸控制台，在左侧导航栏单击 工作流。页面包含 我的工作流、运行记录、工作流模板、子任务模板 四个页签。
2. 切换到 工作流模板 页签，找到 人类演示数据视觉预处理与质量清洗 模板。您可以单击卡片查看模板详情。

3. 单击卡片右下角的 创建工作流，在弹窗中配置以下参数。

   • 工作流名称：仅支持大小写字母、数字、-、_、.，例如 p1-visual-clean-demo。
   • 资源池：选择前提条件中准备的全托管资源池。选择后，控制台会自动校验 aihc-argo-workflows 组件；如提示组件状态异常，请先联系管理员处理。
   • 定时运行：保持关闭。
   • 描述：选填。
4. 单击 确定。页面提示“创建工作流成功”后，自动进入工作流编辑器。

步骤二：获取队列 ID 和 PFS ID

1. 在工作流编辑器中，单击 查看队列信息。
2. 在弹窗中复制 PFS ID，以及要使用的 CPU 队列的队列 ID（格式为 aihcq-xxxx）。弹窗中会显示每个队列的 CPU 和内存空闲量。请确认空闲资源足够运行一个节点（默认 Head 4 核 / 8 GB + Worker 6 核 / 12 GB）。

步骤三：修改全局参数

在工作流编辑器中修改以下全局参数。

配置示例中的 <YOUR-PATH-NAME> 请替换为您在 PFS 上的目录。

在当前线上版本中，您可以任选以下一种方式修改：

• 可视化模式：在配置面板的 总览信息 > 全局输入参数 表单中，填入队列 ID（queue_id）、PFS ID（pfs_id）等参数。

• Yaml 模式：直接修改顶部 inputs 中对应参数的 defaultValue。

  说明

  工作流参数分为全局参数和节点参数：

  • 全局参数：被多个节点共同引用，修改一处即全部生效。例如本例中的 tmp_dir 被 6 个节点共同引用。
  • 节点参数：仅作用于所在节点，例如本例中 step6 的 clarity_min。在可视化模式下，单击节点后在 子任务信息 > 输入参数 中修改；在 Yaml 模式下，修改 tasks 中该节点的 inputs.value。

本例的 6 个节点使用同一套资源参数。在实际管线中，如果某个节点需要 GPU（例如接在 P1 之后的 HaWoR 手部重建），只需在该节点的资源配置中添加卡型和卡数，其他节点不受影响。数据量较大时，调大 worker_replicas 即可横向扩展。

步骤四：保存并运行

1. 单击右上角的 保存。
2. 单击 运行（运行前会自动保存一次）。页面提示“运行成功”后，将跳转到该工作流的运行记录。

步骤五：查看运行状态、日志和监控

1. 在运行记录中单击本次运行，查看 6 个子任务的运行状态、持续时间和结束时间。子任务按 DAG 顺序执行，上一个子任务成功后，下一个才会开始。
2. 在子任务所在行的 操作 列，可以打开该子任务的 监控 和 日志。
3. 6 个子任务全部成功，即表示整条管线已跑通。

步骤六：查看结果

使用挂载同一个 PFS 实例的开发机查看输出结果。开发机的 PFS 挂载配置要求如下，详细操作请参见开发机存储挂载与数据集挂载。

• PFS ID：与工作流全局参数 pfs_id 一致。
• 源路径：/。
• 目标路径：/mnt/pfs。

1. 在开发机终端中执行以下命令，查看输出目录和结果文件。

1ls /mnt/pfs/<YOUR-PATH-NAME>/p1/demo-001/           # clips.jsonl 和 tmp/
2cat /mnt/pfs/<YOUR-PATH-NAME>/p1/demo-001/clips.jsonl

说明

开发机 WebIDE 左侧默认只显示 /root/workspace 目录，看不到 /mnt/pfs。您可以通过菜单 File > Open Folder 打开 /mnt/pfs/<你的目录>/p1，或在终端执行 ln -s /mnt/pfs/<你的目录>/p1 /root/workspace/p1 创建软链接。

2. 查看 clips.jsonl 的内容。文件中每行是一个通过过滤的片段，示例如下：

1   {"clip": "/mnt/pfs/zhangsan/p1/demo-001/tmp/clips/clip_000.mp4", "duration": 2.0, "clarity": 835.5, "motion": 0.4000000059604645}

各字段说明如下。

• clarity 值越大画面越清晰，没有上限，仅适合在同一批数据内部比较。
• motion 越接近 0，画面越接近静止。

中间文件保存在 tmp/ 目录下。完整的统计值（如 clarity 的 median/max/min、motion 的 motion_pattern 等）保存在 s5_motion.jsonl 中。

使用自有数据运行管线

完成样例数据的运行后，您可以将输入替换为自己的视频数据。

准备 manifest 文件

manifest 文件描述待处理的视频清单，每条记录对应一个视频，包含以下字段。

manifest 支持 JSON 数组格式，也支持每行一个对象的 JSONL 格式。JSONL 格式示例如下：

1{"video":"/mnt/pfs/<YOUR-PATH-NAME>/p1/input/videos/clip_000.mp4","intrinsics":[[640.0,0.0,320.0],[0.0,640.0,240.0],[0.0,0.0,1.0]],"distortion":[0.0,0.0,0.0,0.0]}

选择数据存放方式

根据数据存放位置，从以下两种方式中选择一种。

注意

方式二通过 FUSE 将 BOS 映射为目录，使用最简单，但在小文件多、随机读写多的场景下，性能明显低于 PFS。此外，BOSFS 写文件时在 close 后才上传，且不支持回退写；写 MP4 这类需要 seek 的文件时，需先写到本地 /tmp，再拷贝到挂载目录。

方式一：数据放在 PFS

1. 将视频和 manifest 文件上传到 PFS，例如 /mnt/pfs/<YOUR-PATH-NAME>/p1/input/。您可以通过挂载同一 PFS 实例的开发机上传。

2. 修改工作流参数：

   • 将 input_file 修改为 /mnt/pfs/<YOUR-PATH-NAME>/p1/input/manifest.json。
   • 为 tmp_dir、output_file 换一个新的批次号。
   • 如需去畸变，将 no_undistort 修改为 "false"，并在 step1 的节点参数中将 camera_model 设置为 pinhole 或 fisheye。
3. 先用小样本验证。建议先用 5~20 条样本运行一遍，查看 clips.jsonl 中 clarity、motion 的分布，再调整 step6 的三个阈值。确认效果后再处理全量数据；数据量较大时，调大 worker_replicas。

方式二：数据放在 BOS

将您自己的 bucket 挂载到容器中，输入、中间结果和输出都放在 BOS 上，不再使用 PFS。

1. 确认资源池已支持 BOS 挂载（需安装 BOSFS），详情请参见在训练任务中使用云存储。
2. 在 Yaml 模式中，将每个节点（6 个 taskTemplates）的 datasources 修改为以下写法：保留平台公共 BOS（其中存放管线脚本），将 PFS 替换为您自己的 bucket。

1       datasources:
2       - type: bos                      # 平台公共 BOS，脚本和样例数据，保持不变
3         name: ""
4         sourcePath: '{{pipeline.parameters.bos_source_path}}'
5         mountPath: '{{pipeline.parameters.bos_mount_path}}'
6       - type: bos                      # 自己的 bucket，替换原来的 PFS 挂载
7         name: ""
8         sourcePath: <your-bucket>/<prefix>
9         mountPath: /mnt/bos

3. 修改工作流参数：将 input_file、tmp_dir、output_file 都修改为 /mnt/bos/... 下的路径。此时 pfs_id 等 PFS 相关参数已不再被引用，可以保留或删除。