基于 Lance 宽表的视频全链路精选与多格式生成管线

更新时间：2026-08-06

概述

此管线基于 Lance 列式表为核心数据面，对海量原始视频做全链路质量精选：从场景切分、语义去重、美学评分、NSFW 过滤到结构化 Caption，最终输出训练可用的 WebDataset 分片和 LeRobot v3.0 数据集。

输入：原始视频/图像清单（manifest.jsonl，含 sample_id / modality / url）
输出：
- WebDataset tar 分片（供视觉-语言训练）
- LeRobot v3.0 数据集（供具身策略训练）
业务价值：将散乱原始素材一站式精选为高质量训练数据，前置 CPU/GPU 联合清洗显著降低下游训练噪声

处理流程

Plain Text

1manifest.jsonl → S00 写入Lance → S01 校验入库 → S02 场景切分 → S03 视频嵌入(GPU)
2→ S04 KMeans聚类 → S05 语义去重 → S06 美学评分(GPU) → S07 NSFW检测(GPU)
3→ S08 质量过滤 → S09 结构化Caption(GPU) → S10 WebDataset分片 / S11 导出LeRobot v3.0

算子映射

步骤	算子	GPU	说明
S00	`IngestToLance`	否	从 manifest.jsonl 写入 Lance 宽表
S01	`Ingest`	否	字段校验 + 默认值填充
S02	`VideoSplitByScene`	否	ContentDetector 场景切分，1→N 行 explode
S03	`CosmosVideoEmbedding`	是	Cosmos-Embed1-448p 视频嵌入
S04	`KMeansClustering`	否	嵌入向量 KMeans 聚类
S05	`SemanticDedupByCosine`	否	余弦相似度语义去重
S06	`DOVERAesthetic`	是	DOVER 视频美学评分
S07	`VideoNsfwDetect`	是	Falconsai NSFW 检测
S08	Filter	否	按去重/美学/NSFW 三维过滤
S09	`StructuredCaptionPipeline`	是	Qwen2.5-VL 结构化 Caption
S10	`ExportToWebDataset`	否	输出 WebDataset tar 分片
S11	`ExportLeRobotV30`	否	输出 LeRobot v3.0 数据集

数据流向

Plain Text

1manifest.jsonl
2  → S00 → lance://bos://<YOUR-BUCKET-NAME>/aihc/p6-run1/episodes.lance
3  → S01 → (Lance 表内 keep=True 校验)
4  → S02 → (Lance 表追加 clip 列，行 explode)
5  → S03 → (Lance 表追加 video_embed 列)
6  → S04 → (Lance 表追加 cluster_id 列)
7  → S05 → (Lance 表追加 dedup_keep 列)
8  → S06 → (Lance 表追加 dover_aesthetic 列)
9  → S07 → (Lance 表追加 video_nsfw 列)
10  → S08 → (Lance 表更新 keep + drop_reasons)
11  → S09 → (Lance 表追加 caption 列)
12  → S10 → output_dir/*.tar (WebDataset 分片)
13  → S11 → output_dir/ (LeRobot v3.0 数据集)

前置条件

资源	说明
AIHC 队列	需要 CPU 队列和 GPU 队列（分别用于 CPU/GPU 步骤）
BOS Bucket	存放 Lance 表、输入数据和模型（默认 `storage-aihc,需提前创建`）
GPU 资源	S03/S06/S07 需要 RTX 4090；S09 需要 2×RTX 4090
镜像	`ccr-registry.baidubce.com/aihc/aihc-daft-gpu:0.5.2.12.2-cu129-py3.12-p6-sm120-sm89-ubuntu22.04`
模型	Cosmos-Embed1-448p、DOVER、Falconsai/nsfw_image_detection、Qwen2.5-VL-7B-Instruct（镜像内 `/opt/aihc/models`）

全局参数说明

参数名	类型	必填	默认值	说明
`image`	string	是	`ccr-registry.baidubce.com/aihc/aihc-daft-gpu:0.5.2.12.2-cu129-py3.12-p6-sm120-sm89-ubuntu22.04`	镜像地址
`cpu_queue_id`	string	是	—	CPU 计算队列 ID
`gpu_queue_id`	string	是	—	GPU 计算队列 ID
`priority`	string	否	`high`	任务优先级
`lance_uri`	string	是	`lance://bos://storage-aihc/aihc/p6-run1/episodes.lance`	Lance 表 URI，需要替换为自己的 BOS 地址
`manifest`	string	是	—	输入 manifest.jsonl 路径（BOS 上）
`output_dir`	string	是	—	最终输出目录（WebDataset 分片 + LeRobot 数据集）
`bos_env_file`	string	是	-	BOS 凭证文件路径（需要包含 ACCESS_KEY/SECRET_KEY，要求可以读写`lance_uri`参数中的填写的BOS地址）,lance 数据的读写基于 bos endpoint，填写示例见下方
`source`	string	否	`manifest`	数据来源标签
`source_dataset`	string	否	manifest 父目录名	数据集标签
`head_cpu`	int	否	`6`	Head 节点 CPU 数
`head_memory`	int	否	`12`	Head 节点内存（GB）
`worker_replicas`	int	否	`2`	Worker 副本数
`worker_cpu`	int	否	`16`	Worker 节点 CPU 数
`worker_memory`	int	否	`64`	Worker 节点内存（GB）
`worker_gpu`	int	否	`1`	Worker 节点 GPU 数（GPU 步骤使用）

bos_env_file文件内容要求：

Plain Text

1export BOS_ENDPOINT=https://bj.bcebos.com
2export BOS_S3_ENDPOINT=https://s3.bj.bcebos.com
3export BOS_REGION=bj
4export BOS_ACCESS_KEY=xxxxxxx
5export BOS_SECRET_KEY=xxxxxxx

步骤详解

S00 写入 Lance（Ingest to Lance）

从 manifest.jsonl 读取记录，写入 Lance 宽表作为后续所有步骤的统一数据面。支持增量追加（按 sample_id 去重），--force 可全量覆盖。

参数	类型	默认值	说明
`manifest`	string	—	输入 manifest.jsonl 路径
`output`	string	—	Lance URI
`source`	string	`manifest`	来源标签
`source_dataset`	string	父目录名	数据集标签
`force`	bool	`false`	是否强制覆盖

输入：manifest.jsonl（字段：sample_id, modality, url）
输出：Lance 表（schema：sample_id, modality, url, source, source_dataset, ingest_time, keep, meta）
资源：CPU-only，纯 IO 操作

S01 字段校验（Ingest）

校验 manifest.jsonl 字段合法性，跳过缺失 sample_id / 非法 modality / 无 url 的行，为合法行填充 keep=True 默认值。

参数	类型	默认值	说明
`allow_empty`	bool	`false`	允许空输入不报错

输入：manifest.jsonl
输出：校验后的 JSONL
资源：CPU-only，纯 IO 操作

S02 场景切分（Split Scene）

使用 PySceneDetect ContentDetector 检测画面内容突变点，将长视频切分为独立场景片段。切分后行 explode，每行对应一个 clip。仅处理 modality=video 的行。

参数	类型	默认值	说明
`detector`	string	`ContentDetector`	场景检测算法
`threshold`	float	`27.0`	检测灵敏度阈值（越小越敏感）
`min_scene_len`	int	`15`	最小场景帧数

输入列：url（视频路径）
输出列：新增 clip（切分后的片段路径）
资源：CPU-only（4 CPU / 0 GPU）

S03 视频嵌入（Embed Video）-GPU

使用 NVIDIA Cosmos-Embed1-448p 模型生成视频语义嵌入向量，供后续聚类和去重使用。

参数	类型	默认值	说明
`model_name`	string	`nvidia/Cosmos-Embed1-448p`	嵌入模型标识
`model_path`	string	`/opt/aihc/models`	模型根目录
`model_batch_size`	int	`4`	GPU forward batch size
`num_frames`	int	`8`	每段视频采样帧数

输入列：视频路径（优先级：compressed > cropped > clip > url）
输出列：新增 video_embed（嵌入向量）
资源：1×RTX 4090，2 CPU，16 CPU / 64GB 内存

S04 KMeans 聚类（KMeans）

对嵌入向量做 KMeans 聚类，分配 cluster_id，为语义去重提供分组依据。

参数	类型	默认值	说明
`n_clusters`	int	`1024`	聚类数
`backend`	string	`auto`	聚类后端（auto/cuml/faiss/sklearn）
`max_iter`	int	`20`	最大迭代数

输入列：video_embed 或 image_embed
输出列：新增 cluster_id（-1 表示无嵌入）
资源：CPU-only，全局单次操作（concurrency=1）

S05 语义去重（Dedup）

在每个 cluster 内按余弦相似度去重，标记重复样本。

参数	类型	默认值	说明
`threshold`	float	`0.95`	余弦相似度去重阈值

输入列：embed, cluster_id
输出列：新增 dedup_keep（bool，True=保留）
资源：CPU-only，全局单次操作

S06 美学评分（Dover Aesthetic）- GPU

使用 DOVER 模型对视频进行美学质量评分。

参数	类型	默认值	说明
`backend`	string	`dover`	评分后端（dover/laion）
`model_path`	string	`/opt/aihc/model/dover`	DOVER 权重目录
`num_frames`	int	`32`	采样帧数
`num_clips`	int	`1`	采样 clip 数

输入列：视频路径
输出列：新增 dover_aesthetic（float 评分）
资源：0.5×RTX 4090，2 CPU；Workers 3×(16 CPU / 64GB)

S07 NSFW 检测（NSFW Video）- GPU

使用 Falconsai/nsfw_image_detection 模型检测视频 NSFW 概率。

参数	类型	默认值	说明
`model_name`	string	`Falconsai/nsfw_image_detection`	模型标识
`model_path`	string	`/opt/aihc/models`	模型根目录
`num_frames`	int	`8`	均匀采样帧数

输入列：视频路径
输出列：新增 video_nsfw（概率 float）
资源：0.25×RTX 4090，2 CPU；Workers 3×(16 CPU / 64GB)

S08 质量过滤（Filter）

根据去重、美学评分、NSFW 检测三个维度综合过滤，更新 keep 标记。

参数	类型	默认值	说明
`dover_aesthetic_min`	float	`3.0`	美学评分最低阈值
`nsfw_threshold`	float	`0.5`	NSFW 概率上限

输入列：dedup_keep, dover_aesthetic, video_nsfw
输出列：更新 keep（bool）+ 新增 drop_reasons（列表）
资源：CPU-only，纯 pandas/pyarrow 操作
过滤规则：
- dedup_keep=False → drop_reason: "dedup"
- dover_aesthetic < dover_aesthetic_min → drop_reason: "dover_aes_low"
- video_nsfw > nsfw_threshold → drop_reason: "video_nsfw"

S09 结构化 Caption（Caption）- GPU

使用 Qwen2.5-VL-7B-Instruct 通过 vLLM 对保留视频生成结构化文本描述。支持视频和图像双分支。

参数	类型	默认值	说明
`model_name`	string	`Qwen/Qwen2.5-VL-7B-Instruct`	模型标识
`model_path`	string	`/opt/aihc/models`	模型根目录
`backend`	string	`auto`	推理后端（auto/vllm/transformers）
`tensor_parallel_size`	int	`-1`（自动）	张量并行 GPU 数
`gpu_memory_utilization`	float	`0.85`	vLLM 显存占比
`max_model_len`	int	`-1`（自动）	最大序列长度
`enforce_eager`	bool	`false`	禁用 CUDA graphs
`lora_path`	string	空	LoRA 适配器路径（可选）

输入列：keep=True 的视频/图像路径
输出列：新增 caption（结构化文本）
资源：2×RTX 4090，16 CPU / 128GB 内存

S10 WebDataset 分片（Shard）

将 keep=True 的样本导出为 WebDataset tar 分片。

参数	类型	默认值	说明
`output_dir`	string	—	分片输出目录
`shard_prefix`	string	`shard`	tar 文件前缀
`max_shard_size_mb`	int	`512`	单个分片最大 MB
`max_samples_per_shard`	int	`10000`	单个分片最大样本数

输入：Lance 表中 keep=True 的行
输出：output_dir/shard-000000.tar 等
资源：CPU-only，全局单次有序操作

S11 导出 LeRobot v3.0（Export LeRobot）

将 keep=True 的具身数据（含 action/state）导出为 LeRobot v3.0 格式数据集。

参数	类型	默认值	说明
`out`	string	—	LeRobot v3.0 输出目录
`repo_id`	string	`p6_curated_v3`	数据集 repo ID
`fps`	int	`30`	视频帧率
`robot_type`	string	`generic_hand`	机器人类型标签
`action_dim`	int	`7`	动作维度
`use_videos`	bool	`false`	是否包含视频列
`vcodec`	string	`h264`	视频编码器

输入：Lance 表中 keep=True 且含 action/state 的行
输出：LeRobot v3.0 数据集目录（data/*.parquet）
资源：CPU-only，Workers 2×(16 CPU / 64GB)

任务拓扑

[流程图 7342279715ae42a68dfbc53857a9b751 mxGraph]

S00 → S09：严格串行，每步依赖前一步
S09 → S10 / S11：并行执行（两个输出分支互不依赖）
GPU 步骤：S03、S06、S07、S09

完整工作流YAML

                Python
                
            

                version: v1
kind: PipelineTemplate

inputs:
- name: image
  type: string
  hint: 镜像地址
  defaultValue: ccr-registry.baidubce.com/aihc/aihc-daft-gpu:0.5.2.12.2-cu129-py3.12-p6-sm120-sm89-ubuntu22.04
- name: cpu_queue_id
  type: string
  hint: CPU 队列 ID
  defaultValue: <YOUR_QUEUE_ID>
- name: gpu_queue_id
  type: string
  hint: GPU 队列 ID
  defaultValue: <YOUR_QUEUE_ID>
- name: priority
  type: string
  defaultValue: high
- name: pfs_id
  type: string
  hint: PFS 实例 ID
  defaultValue: <YOUR_PFS_ID>
- name: pfs_source_path
  type: string
  defaultValue: /
- name: pfs_mount_path
  type: string
  defaultValue: /mnt/pfs
- name: bos_source_path
  type: string
  hint: BOS sourcePath（代码/模型）
  defaultValue: aihc-rdw-bj/aihc-daft
- name: bos_mount_path
  type: string
  hint: BOS 挂载点（代码/模型）
  defaultValue: /opt/aihc
- name: input
  type: string
  hint: 源视频 manifest.jsonl 路径
  defaultValue: /mnt/pfs/<YOUR_PATH>/input_data/p6/manifest.jsonl
- name: lance_uri
  type: string
  hint: Lance 宽表 URI（s01-s10 共用）
  defaultValue: bos://storage-aihc/aihc/<YOUR_PATH>/episodes.lance
- name: tmp_dir
  type: string
  hint: 中间数据目录
  defaultValue: /mnt/pfs/<YOUR_PATH>/tmp_data/p6
- name: output_dir
  type: string
  hint: 最终产物输出目录
  defaultValue: /mnt/pfs/<YOUR_PATH>/output_data/p6
# BOS 凭证
- name: bos_env_file
  type: string
  hint: BOS 凭证文件路径（含 BOS_ACCESS_KEY/BOS_SECRET_KEY）
  defaultValue: /mnt/pfs/<YOUR_PATH>/env/bos_env
# S00 IngestToLance
- name: s00_source
  type: string
  defaultValue: manifest
- name: s00_force
  type: string
  hint: 传 "--force" 则强制覆盖 Lance 表，留空则增量 append
  defaultValue: "--force"
# S02 SplitScene
- name: threshold
  type: string
  defaultValue: "27.0"
- name: min_scene_len
  type: string
  defaultValue: "15"
- name: s02_num_cpus
  type: string
  defaultValue: "4.0"
- name: s02_worker_replicas
  type: int
  defaultValue: 4
- name: s02_worker_cpu
  type: int
  defaultValue: 16
- name: s02_worker_memory
  type: int
  defaultValue: 64
# S03 EmbedVideo
- name: embed_model_name
  type: string
  defaultValue: /opt/aihc/model/nvidia/Cosmos-Embed1-448p
- name: s03_num_gpus
  type: string
  defaultValue: "1.0"
- name: s03_worker_replicas
  type: int
  defaultValue: 2
- name: s03_worker_gpu
  type: int
  defaultValue: 1
# S04 KMeans
- name: n_clusters
  type: string
  defaultValue: "1024"
# S06 DOVER
- name: dover_model_name
  type: string
  defaultValue: /opt/aihc/model/dover
- name: dover_backend
  type: string
  defaultValue: dover
- name: dover_laion_mlp_path
  type: string
  defaultValue: /opt/aihc/model/laion_aesthetic_v2
- name: dover_clip_model
  type: string
  defaultValue: /opt/aihc/model/openai/clip-vit-large-patch14
- name: dover_num_frames
  type: string
  defaultValue: "32"
- name: dover_num_clips
  type: string
  defaultValue: "1"
# S07 NSFW
- name: nsfw_model_name
  type: string
  defaultValue: /opt/aihc/model/Falconsai/nsfw_image_detection
# S08 Filter
- name: dover_aesthetic_min
  type: string
  defaultValue: "3.0"
- name: nsfw_threshold
  type: string
  defaultValue: "0.5"
# S09 Caption
- name: caption_model_name
  type: string
  defaultValue: /opt/aihc/model/Qwen2.5-VL-7B-Instruct
- name: s09_num_gpus
  type: string
  defaultValue: "2.0"
- name: s09_backend
  type: string
  defaultValue: auto
- name: s09_tensor_parallel_size
  type: string
  defaultValue: "-1"
- name: s09_gpu_memory_utilization
  type: string
  defaultValue: "0.85"
- name: s09_max_model_len
  type: string
  defaultValue: "-1"
- name: s09_worker_replicas
  type: int
  defaultValue: 2
- name: s09_worker_cpu
  type: int
  defaultValue: 16
- name: s09_worker_memory
  type: int
  defaultValue: 128
- name: s09_worker_gpu
  type: int
  defaultValue: 2
# S10 Shard
- name: max_shard_size_mb
  type: string
  defaultValue: "512.0"
- name: shard_prefix
  type: string
  defaultValue: shard
- name: max_samples_per_shard
  type: string
  defaultValue: "10000"
# S11 Export LeRobot
- name: repo_id
  type: string
  defaultValue: p6_curated_v3
- name: s11_fps
  type: string
  defaultValue: "30"
- name: s11_robot_type
  type: string
  defaultValue: generic_hand
- name: s11_action_dim
  type: string
  defaultValue: "7"
# 通用资源
- name: head_cpu
  type: int
  defaultValue: 6
- name: head_memory
  type: int
  defaultValue: 12
- name: shared_memory
  type: int
  defaultValue: 30

taskTemplates:
# ========== S00: Ingest to Lance ==========
- name: p6-00-ingest-to-lance
  type: CustomTask
  spec:
    queue: '{{pipeline.parameters.cpu_queue_id}}'
    priority: '{{pipeline.parameters.priority}}'
    jobType: RayJob
    command: >-
      source {{pipeline.parameters.bos_env_file}} &&
      export LD_LIBRARY_PATH=$(python3 -c "import os,sys;print(os.path.join(sys.prefix,'lib'))"):$LD_LIBRARY_PATH &&
      python3 {{pipeline.parameters.bos_mount_path}}/workflow/p6_00_ingest_to_lance.py
      --manifest {{pipeline.parameters.input}}
      --output {{pipeline.parameters.lance_uri}}
      --tmp-dir {{pipeline.parameters.tmp_dir}}
      --source {{pipeline.parameters.s00_source}}
      --num-cpus 1.0
      {{pipeline.parameters.s00_force}}
    jobSpec:
      Head:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.head_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.head_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
      Worker:
        replicas: 2
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: 16
        - name: memory
          quantity: 64
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
    datasources:
    - type: pfs
      sourcePath: '{{pipeline.parameters.pfs_source_path}}'
      mountPath: '{{pipeline.parameters.pfs_mount_path}}'
      name: '{{pipeline.parameters.pfs_id}}'
    - type: bos
      name: ""
      sourcePath: '{{pipeline.parameters.bos_source_path}}'
      mountPath: '{{pipeline.parameters.bos_mount_path}}'
    labels:
      aijob.cce.baidubce.com/create-from-aihcp: "true"
      aijob.cce.baidubce.com/log-collection: v3
    autoCreatePVC: true
    hostNetwork: false
# ========== S01: Ingest ==========
- name: p6-01-ingest
  type: CustomTask
  spec:
    queue: '{{pipeline.parameters.cpu_queue_id}}'
    priority: '{{pipeline.parameters.priority}}'
    jobType: RayJob
    command: >-
      source {{pipeline.parameters.bos_env_file}} &&
      export LD_LIBRARY_PATH=$(python3 -c "import os,sys;print(os.path.join(sys.prefix,'lib'))"):$LD_LIBRARY_PATH &&
      python3 {{pipeline.parameters.bos_mount_path}}/workflow/p6_01_ingest.py
      --input {{pipeline.parameters.lance_uri}}
      --output {{pipeline.parameters.lance_uri}}
      --tmp-dir {{pipeline.parameters.tmp_dir}}
      --num-cpus 1.0
    jobSpec:
      Head:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.head_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.head_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
      Worker:
        replicas: 2
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: 16
        - name: memory
          quantity: 64
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
    datasources:
    - type: pfs
      sourcePath: '{{pipeline.parameters.pfs_source_path}}'
      mountPath: '{{pipeline.parameters.pfs_mount_path}}'
      name: '{{pipeline.parameters.pfs_id}}'
    - type: bos
      name: ""
      sourcePath: '{{pipeline.parameters.bos_source_path}}'
      mountPath: '{{pipeline.parameters.bos_mount_path}}'
    labels:
      aijob.cce.baidubce.com/create-from-aihcp: "true"
      aijob.cce.baidubce.com/log-collection: v3
    autoCreatePVC: true
    hostNetwork: false
# ========== S02: Split Scene (CPU) ==========
- name: p6-02-split-scene
  type: CustomTask
  spec:
    queue: '{{pipeline.parameters.cpu_queue_id}}'
    priority: '{{pipeline.parameters.priority}}'
    jobType: RayJob
    command: >-
      source {{pipeline.parameters.bos_env_file}} &&
      export LD_LIBRARY_PATH=$(python3 -c "import os,sys;print(os.path.join(sys.prefix,'lib'))"):$LD_LIBRARY_PATH &&
      DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p6_02_split_scene.py
      --input {{pipeline.parameters.lance_uri}}
      --output {{pipeline.parameters.lance_uri}}
      --tmp-dir {{pipeline.parameters.tmp_dir}}
      --threshold {{pipeline.parameters.threshold}}
      --min-scene-len {{pipeline.parameters.min_scene_len}}
      --num-cpus {{pipeline.parameters.s02_num_cpus}}
      --batch-size 1
    jobSpec:
      Head:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.head_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.head_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
      Worker:
        replicas: '{{pipeline.parameters.s02_worker_replicas}}'
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.s02_worker_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.s02_worker_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
    datasources:
    - type: pfs
      sourcePath: '{{pipeline.parameters.pfs_source_path}}'
      mountPath: '{{pipeline.parameters.pfs_mount_path}}'
      name: '{{pipeline.parameters.pfs_id}}'
    - type: bos
      name: ""
      sourcePath: '{{pipeline.parameters.bos_source_path}}'
      mountPath: '{{pipeline.parameters.bos_mount_path}}'
    labels:
      aijob.cce.baidubce.com/create-from-aihcp: "true"
      aijob.cce.baidubce.com/log-collection: v3
    autoCreatePVC: true
    hostNetwork: false
# ========== S03: Embed Video (GPU) ==========
- name: p6-03-embed-video
  type: CustomTask
  spec:
    queue: '{{pipeline.parameters.gpu_queue_id}}'
    priority: '{{pipeline.parameters.priority}}'
    jobType: RayJob
    command: >-
      source {{pipeline.parameters.bos_env_file}} &&
      export LD_LIBRARY_PATH=$(python3 -c "import os,sys;print(os.path.join(sys.prefix,'lib'))"):$LD_LIBRARY_PATH &&
      DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p6_03_embed_video.py
      --input {{pipeline.parameters.lance_uri}}
      --output {{pipeline.parameters.lance_uri}}
      --tmp-dir {{pipeline.parameters.tmp_dir}}
      --model-name {{pipeline.parameters.embed_model_name}}
      --num-cpus 2.0
      --num-gpus {{pipeline.parameters.s03_num_gpus}}
      --batch-size 1
    jobSpec:
      Head:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.head_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.head_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        - name: AIHC_MODEL_DIR
          value: "{{pipeline.parameters.embed_model_name}}"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
      Worker:
        replicas: '{{pipeline.parameters.s03_worker_replicas}}'
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: 16
        - name: memory
          quantity: 64
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        - name: baidu.com/rtx_4090_cgpu
          quantity: '{{pipeline.parameters.s03_worker_gpu}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        - name: AIHC_MODEL_DIR
          value: "{{pipeline.parameters.embed_model_name}}"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
    datasources:
    - type: pfs
      sourcePath: '{{pipeline.parameters.pfs_source_path}}'
      mountPath: '{{pipeline.parameters.pfs_mount_path}}'
      name: '{{pipeline.parameters.pfs_id}}'
    - type: bos
      name: ""
      sourcePath: '{{pipeline.parameters.bos_source_path}}'
      mountPath: '{{pipeline.parameters.bos_mount_path}}'
    labels:
      aijob.cce.baidubce.com/create-from-aihcp: "true"
      aijob.cce.baidubce.com/log-collection: v3
    autoCreatePVC: true
    hostNetwork: false
# ========== S04: KMeans (CPU) ==========
- name: p6-04-kmeans
  type: CustomTask
  spec:
    queue: '{{pipeline.parameters.cpu_queue_id}}'
    priority: '{{pipeline.parameters.priority}}'
    jobType: RayJob
    command: >-
      source {{pipeline.parameters.bos_env_file}} &&
      export LD_LIBRARY_PATH=$(python3 -c "import os,sys;print(os.path.join(sys.prefix,'lib'))"):$LD_LIBRARY_PATH &&
      DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p6_04_kmeans.py
      --input {{pipeline.parameters.lance_uri}}
      --output {{pipeline.parameters.lance_uri}}
      --tmp-dir {{pipeline.parameters.tmp_dir}}
      --n-clusters {{pipeline.parameters.n_clusters}}
      --num-cpus 4.0
    jobSpec:
      Head:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.head_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.head_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
      Worker:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: 16
        - name: memory
          quantity: 64
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
    datasources:
    - type: pfs
      sourcePath: '{{pipeline.parameters.pfs_source_path}}'
      mountPath: '{{pipeline.parameters.pfs_mount_path}}'
      name: '{{pipeline.parameters.pfs_id}}'
    - type: bos
      name: ""
      sourcePath: '{{pipeline.parameters.bos_source_path}}'
      mountPath: '{{pipeline.parameters.bos_mount_path}}'
    labels:
      aijob.cce.baidubce.com/create-from-aihcp: "true"
      aijob.cce.baidubce.com/log-collection: v3
    autoCreatePVC: true
    hostNetwork: false
# ========== S05: Dedup (CPU) ==========
- name: p6-05-dedup
  type: CustomTask
  spec:
    queue: '{{pipeline.parameters.cpu_queue_id}}'
    priority: '{{pipeline.parameters.priority}}'
    jobType: RayJob
    command: >-
      source {{pipeline.parameters.bos_env_file}} &&
      export LD_LIBRARY_PATH=$(python3 -c "import os,sys;print(os.path.join(sys.prefix,'lib'))"):$LD_LIBRARY_PATH &&
      DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p6_05_dedup.py
      --input {{pipeline.parameters.lance_uri}}
      --output {{pipeline.parameters.lance_uri}}
      --tmp-dir {{pipeline.parameters.tmp_dir}}
      --threshold 0.95
      --num-cpus 4.0
    jobSpec:
      Head:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.head_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.head_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
      Worker:
        replicas: 3
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: 16
        - name: memory
          quantity: 64
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
    datasources:
    - type: pfs
      sourcePath: '{{pipeline.parameters.pfs_source_path}}'
      mountPath: '{{pipeline.parameters.pfs_mount_path}}'
      name: '{{pipeline.parameters.pfs_id}}'
    - type: bos
      name: ""
      sourcePath: '{{pipeline.parameters.bos_source_path}}'
      mountPath: '{{pipeline.parameters.bos_mount_path}}'
    labels:
      aijob.cce.baidubce.com/create-from-aihcp: "true"
      aijob.cce.baidubce.com/log-collection: v3
    autoCreatePVC: true
    hostNetwork: false
# ========== S06: DOVER Aesthetic (GPU) ==========
- name: p6-06-dover-aesthetic
  type: CustomTask
  spec:
    queue: '{{pipeline.parameters.gpu_queue_id}}'
    priority: '{{pipeline.parameters.priority}}'
    jobType: RayJob
    command: >-
      source {{pipeline.parameters.bos_env_file}} &&
      export LD_LIBRARY_PATH=$(python3 -c "import os,sys;print(os.path.join(sys.prefix,'lib'))"):$LD_LIBRARY_PATH &&
      DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p6_06_dover_aesthetic.py
      --input {{pipeline.parameters.lance_uri}}
      --output {{pipeline.parameters.lance_uri}}
      --tmp-dir {{pipeline.parameters.tmp_dir}}
      --model-path {{pipeline.parameters.dover_model_name}}
      --backend {{pipeline.parameters.dover_backend}}
      --laion-mlp-path {{pipeline.parameters.dover_laion_mlp_path}}
      --clip-model {{pipeline.parameters.dover_clip_model}}
      --num-frames {{pipeline.parameters.dover_num_frames}}
      --num-clips {{pipeline.parameters.dover_num_clips}}
      --num-cpus 2.0
      --num-gpus 0.5
      --batch-size 1
    jobSpec:
      Head:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.head_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.head_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        - name: AIHC_MODEL_DIR
          value: "{{pipeline.parameters.dover_model_name}}"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
      Worker:
        replicas: 3
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: 16
        - name: memory
          quantity: 64
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        - name: baidu.com/rtx_4090_cgpu
          quantity: 1
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        - name: AIHC_MODEL_DIR
          value: "{{pipeline.parameters.dover_model_name}}"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
    datasources:
    - type: pfs
      sourcePath: '{{pipeline.parameters.pfs_source_path}}'
      mountPath: '{{pipeline.parameters.pfs_mount_path}}'
      name: '{{pipeline.parameters.pfs_id}}'
    - type: bos
      name: ""
      sourcePath: '{{pipeline.parameters.bos_source_path}}'
      mountPath: '{{pipeline.parameters.bos_mount_path}}'
    labels:
      aijob.cce.baidubce.com/create-from-aihcp: "true"
      aijob.cce.baidubce.com/log-collection: v3
    autoCreatePVC: true
    hostNetwork: false
# ========== S07: NSFW Video (GPU) ==========
- name: p6-07-nsfw-video
  type: CustomTask
  spec:
    queue: '{{pipeline.parameters.gpu_queue_id}}'
    priority: '{{pipeline.parameters.priority}}'
    jobType: RayJob
    command: >-
      source {{pipeline.parameters.bos_env_file}} &&
      export LD_LIBRARY_PATH=$(python3 -c "import os,sys;print(os.path.join(sys.prefix,'lib'))"):$LD_LIBRARY_PATH &&
      DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p6_07_nsfw_video.py
      --input {{pipeline.parameters.lance_uri}}
      --output {{pipeline.parameters.lance_uri}}
      --tmp-dir {{pipeline.parameters.tmp_dir}}
      --model-name {{pipeline.parameters.nsfw_model_name}}
      --num-cpus 2.0
      --num-gpus 0.25
      --batch-size 1
    jobSpec:
      Head:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.head_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.head_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        - name: AIHC_MODEL_DIR
          value: "{{pipeline.parameters.nsfw_model_name}}"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
      Worker:
        replicas: 3
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: 16
        - name: memory
          quantity: 64
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        - name: baidu.com/rtx_4090_cgpu
          quantity: 1
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        - name: AIHC_MODEL_DIR
          value: "{{pipeline.parameters.nsfw_model_name}}"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
    datasources:
    - type: pfs
      sourcePath: '{{pipeline.parameters.pfs_source_path}}'
      mountPath: '{{pipeline.parameters.pfs_mount_path}}'
      name: '{{pipeline.parameters.pfs_id}}'
    - type: bos
      name: ""
      sourcePath: '{{pipeline.parameters.bos_source_path}}'
      mountPath: '{{pipeline.parameters.bos_mount_path}}'
    labels:
      aijob.cce.baidubce.com/create-from-aihcp: "true"
      aijob.cce.baidubce.com/log-collection: v3
    autoCreatePVC: true
    hostNetwork: false
# ========== S08: Filter (CPU) ==========
- name: p6-08-filter
  type: CustomTask
  spec:
    queue: '{{pipeline.parameters.cpu_queue_id}}'
    priority: '{{pipeline.parameters.priority}}'
    jobType: RayJob
    command: >-
      source {{pipeline.parameters.bos_env_file}} &&
      export LD_LIBRARY_PATH=$(python3 -c "import os,sys;print(os.path.join(sys.prefix,'lib'))"):$LD_LIBRARY_PATH &&
      python3 {{pipeline.parameters.bos_mount_path}}/workflow/p6_08_filter.py
      --input {{pipeline.parameters.lance_uri}}
      --output {{pipeline.parameters.lance_uri}}
      --tmp-dir {{pipeline.parameters.tmp_dir}}
      --dover-aesthetic-min {{pipeline.parameters.dover_aesthetic_min}}
      --nsfw-threshold {{pipeline.parameters.nsfw_threshold}}
      --num-cpus 1.0
    jobSpec:
      Head:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.head_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.head_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
      Worker:
        replicas: 2
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: 16
        - name: memory
          quantity: 64
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
    datasources:
    - type: pfs
      sourcePath: '{{pipeline.parameters.pfs_source_path}}'
      mountPath: '{{pipeline.parameters.pfs_mount_path}}'
      name: '{{pipeline.parameters.pfs_id}}'
    - type: bos
      name: ""
      sourcePath: '{{pipeline.parameters.bos_source_path}}'
      mountPath: '{{pipeline.parameters.bos_mount_path}}'
    labels:
      aijob.cce.baidubce.com/create-from-aihcp: "true"
      aijob.cce.baidubce.com/log-collection: v3
    autoCreatePVC: true
    hostNetwork: false
# ========== S09: Caption (GPU) ==========
- name: p6-09-caption
  type: CustomTask
  spec:
    queue: '{{pipeline.parameters.gpu_queue_id}}'
    priority: '{{pipeline.parameters.priority}}'
    jobType: RayJob
    command: >-
      source {{pipeline.parameters.bos_env_file}} &&
      export LD_LIBRARY_PATH=$(python3 -c "import os,sys;print(os.path.join(sys.prefix,'lib'))"):$LD_LIBRARY_PATH &&
      DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p6_09_caption.py
      --input {{pipeline.parameters.lance_uri}}
      --output {{pipeline.parameters.lance_uri}}
      --tmp-dir {{pipeline.parameters.tmp_dir}}
      --model-name {{pipeline.parameters.caption_model_name}}
      --backend {{pipeline.parameters.s09_backend}}
      --tensor-parallel-size {{pipeline.parameters.s09_tensor_parallel_size}}
      --gpu-memory-utilization {{pipeline.parameters.s09_gpu_memory_utilization}}
      --max-model-len {{pipeline.parameters.s09_max_model_len}}
      --num-cpus 2.0
      --num-gpus {{pipeline.parameters.s09_num_gpus}}
      --concurrency 1
      --batch-size 1
    jobSpec:
      Head:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.head_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.head_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        - name: AIHC_MODEL_DIR
          value: "{{pipeline.parameters.caption_model_name}}"
        - name: CUDA_VISIBLE_DEVICES
          value: "0,1"
        - name: VLM_NUM_GPUS
          value: "2"
        - name: VLM_CONCURRENCY
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
      Worker:
        replicas: '{{pipeline.parameters.s09_worker_replicas}}'
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.s09_worker_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.s09_worker_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        - name: baidu.com/rtx_4090_cgpu
          quantity: '{{pipeline.parameters.s09_worker_gpu}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        - name: CUDA_VISIBLE_DEVICES
          value: "0,1"
        - name: VLM_NUM_GPUS
          value: "2"
        - name: VLM_CONCURRENCY
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
    datasources:
    - type: pfs
      sourcePath: '{{pipeline.parameters.pfs_source_path}}'
      mountPath: '{{pipeline.parameters.pfs_mount_path}}'
      name: '{{pipeline.parameters.pfs_id}}'
    - type: bos
      name: ""
      sourcePath: '{{pipeline.parameters.bos_source_path}}'
      mountPath: '{{pipeline.parameters.bos_mount_path}}'
    labels:
      aijob.cce.baidubce.com/create-from-aihcp: "true"
      aijob.cce.baidubce.com/log-collection: v3
    autoCreatePVC: true
    hostNetwork: false
# ========== S10: Shard (CPU) ==========
- name: p6-10-shard
  type: CustomTask
  spec:
    queue: '{{pipeline.parameters.cpu_queue_id}}'
    priority: '{{pipeline.parameters.priority}}'
    jobType: RayJob
    command: >-
      source {{pipeline.parameters.bos_env_file}} &&
      export LD_LIBRARY_PATH=$(python3 -c "import os,sys;print(os.path.join(sys.prefix,'lib'))"):$LD_LIBRARY_PATH &&
      DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p6_10_shard.py
      --input {{pipeline.parameters.lance_uri}}
      --output {{pipeline.parameters.lance_uri}}
      --tmp-dir {{pipeline.parameters.tmp_dir}}
      --output-dir {{pipeline.parameters.output_dir}}/shards
      --shard-prefix {{pipeline.parameters.shard_prefix}}
      --max-shard-size-mb {{pipeline.parameters.max_shard_size_mb}}
      --max-samples-per-shard {{pipeline.parameters.max_samples_per_shard}}
      --num-cpus 2.0
    jobSpec:
      Head:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.head_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.head_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
      Worker:
        replicas: 0
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: 16
        - name: memory
          quantity: 64
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
    datasources:
    - type: pfs
      sourcePath: '{{pipeline.parameters.pfs_source_path}}'
      mountPath: '{{pipeline.parameters.pfs_mount_path}}'
      name: '{{pipeline.parameters.pfs_id}}'
    - type: bos
      name: ""
      sourcePath: '{{pipeline.parameters.bos_source_path}}'
      mountPath: '{{pipeline.parameters.bos_mount_path}}'
    labels:
      aijob.cce.baidubce.com/create-from-aihcp: "true"
      aijob.cce.baidubce.com/log-collection: v3
    autoCreatePVC: true
    hostNetwork: false
# ========== S11: Export LeRobot v3.0 (CPU) ==========
- name: p6-11-export-lerobot-v30
  type: CustomTask
  spec:
    queue: '{{pipeline.parameters.cpu_queue_id}}'
    priority: '{{pipeline.parameters.priority}}'
    jobType: RayJob
    command: >-
      source {{pipeline.parameters.bos_env_file}} &&
      export LD_LIBRARY_PATH=$(python3 -c "import os,sys;print(os.path.join(sys.prefix,'lib'))"):$LD_LIBRARY_PATH &&
      DAFT_RUNNER=ray python3 {{pipeline.parameters.bos_mount_path}}/workflow/p6_11_export_lerobot_v30.py
      --input {{pipeline.parameters.lance_uri}}
      --output {{pipeline.parameters.lance_uri}}
      --tmp-dir {{pipeline.parameters.tmp_dir}}
      --out {{pipeline.parameters.output_dir}}/lerobot_v3
      --repo-id {{pipeline.parameters.repo_id}}
      --fps {{pipeline.parameters.s11_fps}}
      --robot-type {{pipeline.parameters.s11_robot_type}}
      --action-dim {{pipeline.parameters.s11_action_dim}}
      --num-cpus 1.0
      --concurrency 1
      --batch-size 1
    jobSpec:
      Head:
        replicas: 1
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: '{{pipeline.parameters.head_cpu}}'
        - name: memory
          quantity: '{{pipeline.parameters.head_memory}}'
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
      Worker:
        replicas: 2
        restartPolicy: Never
        image: '{{pipeline.parameters.image}}'
        resources:
        - name: cpu
          quantity: 16
        - name: memory
          quantity: 64
        - name: sharedMemory
          quantity: '{{pipeline.parameters.shared_memory}}'
        cardLabel: ""
        envs:
        - name: LOG_COLLECTION
          value: "true"
        - name: DAFT_RUNNER
          value: "ray"
        - name: BOS_ENDPOINT
          value: "https://bj.bcebos.com"
        - name: BOS_S3_ENDPOINT
          value: "https://s3.bj.bcebos.com"
        - name: BOS_REGION
          value: "bj"
        - name: LANCE_DATA_STORAGE_VERSION
          value: "stable"
        - name: LANCE_ENABLE_V2_MANIFEST_PATHS
          value: "1"
        - name: HF_HUB_OFFLINE
          value: "1"
        - name: TRANSFORMERS_OFFLINE
          value: "1"
        command: ""
        args: ""
        postStart: ""
        preStop: ""
    datasources:
    - type: pfs
      sourcePath: '{{pipeline.parameters.pfs_source_path}}'
      mountPath: '{{pipeline.parameters.pfs_mount_path}}'
      name: '{{pipeline.parameters.pfs_id}}'
    - type: bos
      name: ""
      sourcePath: '{{pipeline.parameters.bos_source_path}}'
      mountPath: '{{pipeline.parameters.bos_mount_path}}'
    labels:
      aijob.cce.baidubce.com/create-from-aihcp: "true"
      aijob.cce.baidubce.com/log-collection: v3
    autoCreatePVC: true
    hostNetwork: false

# ========== DAG: 线性 12 步 ==========
tasks:
- name: step00-ingest-to-lance
  taskTemplateName: p6-00-ingest-to-lance

- name: step01-ingest
  taskTemplateName: p6-01-ingest
  dependencies:
  - step00-ingest-to-lance

- name: step02-split-scene
  taskTemplateName: p6-02-split-scene
  dependencies:
  - step01-ingest

- name: step03-embed-video
  taskTemplateName: p6-03-embed-video
  dependencies:
  - step02-split-scene

- name: step04-kmeans
  taskTemplateName: p6-04-kmeans
  dependencies:
  - step03-embed-video

- name: step05-dedup
  taskTemplateName: p6-05-dedup
  dependencies:
  - step04-kmeans

- name: step06-dover-aesthetic
  taskTemplateName: p6-06-dover-aesthetic
  dependencies:
  - step05-dedup

- name: step07-nsfw-video
  taskTemplateName: p6-07-nsfw-video
  dependencies:
  - step06-dover-aesthetic

- name: step08-filter
  taskTemplateName: p6-08-filter
  dependencies:
  - step07-nsfw-video

- name: step09-caption
  taskTemplateName: p6-09-caption
  dependencies:
  - step08-filter

- name: step10-shard
  taskTemplateName: p6-10-shard
  dependencies:
  - step09-caption

- name: step11-export-lerobot
  taskTemplateName: p6-11-export-lerobot-v30
  dependencies:
  - step09-caption
            

单步独立执行

每个步骤也提供独立的工作流模板，可以单独提交运行，适用于调试或从中间步骤恢复执行。

例如单独运行 S03 视频嵌入：

YAML

1version: v1
2kind: PipelineTemplate
3
4inputs:
5- name: image
6  type: string
7  defaultValue: ccr-registry.baidubce.com/aihc/aihc-daft-gpu:0.5.2.12.2-cu129-py3.12-p6-sm120-sm89-ubuntu22.04
8- name: queue_id
9  type: string
10  hint: GPU 队列 ID
11- name: lance_uri
12  type: string
13  hint: Lance 表 URI
14- name: model_name
15  type: string
16  defaultValue: nvidia/Cosmos-Embed1-448p
17- name: num_frames
18  type: int
19  defaultValue: 8
20- name: model_batch_size
21  type: int
22  defaultValue: 4
23
24taskTemplates:
25- name: p6-03-embed-video
26  type: CustomTask
27  spec:
28    queue: '{{pipeline.parameters.queue_id}}'
29    priority: high
30    jobType: RayJob
31    command: >-
32      DAFT_RUNNER=ray python3 /opt/aihc/workflow/p6_03_embed_video.py
33      --input {{pipeline.parameters.lance_uri}}
34      --output {{pipeline.parameters.lance_uri}}
35      --model-name {{pipeline.parameters.model_name}}
36      --num-frames {{pipeline.parameters.num_frames}}
37      --model-batch-size {{pipeline.parameters.model_batch_size}}
38    jobSpec:
39      Head:
40        replicas: 1
41        restartPolicy: Never
42        image: '{{pipeline.parameters.image}}'
43        resources:
44        - name: cpu
45          quantity: 6
46        - name: memory
47          quantity: 12
48      Worker:
49        replicas: 2
50        restartPolicy: Never
51        image: '{{pipeline.parameters.image}}'
52        resources:
53        - name: cpu
54          quantity: 16
55        - name: memory
56          quantity: 64
57        - name: baidu.com/rtx_4090_cgpu
58          quantity: 1
59
60tasks:
61- name: embed-video-task
62  taskTemplateName: p6-03-embed-video

使用建议

Lance 增量处理：S00 支持增量追加（按 sample_id diff），无需每次全量重跑；--force 可强制覆盖重建
GPU 步骤是瓶颈：S03/S06/S07/S09 需要 GPU，可通过增加 worker_replicas 提升吞吐
S09 Caption 资源较重：默认需要 2×RTX 4090 + 128GB 内存，如显存不足可降低 gpu_memory_utilization 或使用更小模型
过滤阈值可调：dover_aesthetic_min 和 nsfw_threshold 可根据数据质量灵活调整
S10/S11 并行输出：WebDataset 分片供视觉-语言训练，LeRobot 供具身策略训练，按需启用
离线模式：镜像内模型预装在 /opt/aihc/models，设置 HF_HUB_OFFLINE=1 避免网络依赖
全局单次操作：S04（KMeans）和 S05（Dedup）为全局操作，concurrency 和 batch-size 参数无效

评价此篇文章

有帮助没帮助

百度智能云

百度百舸 · AI计算平台

百度百舸 · AI计算平台

基于 Lance 宽表的视频全链路精选与多格式生成管线

概述

处理流程

算子映射

数据流向

前置条件

全局参数说明

步骤详解

S00 写入 Lance（Ingest to Lance）

S01 字段校验（Ingest）

S02 场景切分（Split Scene）

S03 视频嵌入（Embed Video）-GPU

S04 KMeans 聚类（KMeans）

S05 语义去重（Dedup）

S06 美学评分（Dover Aesthetic）- GPU

S07 NSFW 检测（NSFW Video）- GPU

S08 质量过滤（Filter）

S09 结构化 Caption（Caption）- GPU

S10 WebDataset 分片（Shard）

S11 导出 LeRobot v3.0（Export LeRobot）

任务拓扑

完整工作流YAML

单步独立执行

使用建议