AIHC-Daft算子开发使用指南

更新时间：2026-04-28

数据处理算子可使用百舸提供的 aihc-daft 包进行开发。aihc-daft 是百度 AI 异构计算平台（AIHC）推出的多模态 AI 数据处理框架，基于 Daft 构建，提供开箱即用的数据处理算子库，支持单机多核到多机分布式的弹性扩展，面向 AI 训练数据生产场景。

Daft 核心特性

弹性分布式执行。支持单机多核（Native Runner）与多机集群（Ray Runner）两种执行模式，业务代码无需修改，一行配置即可从单机无缝扩展至百节点集群，满足从开发调试到 PB 级数据生产的全场景需求。
惰性求值与查询优化。采用惰性执行模型，所有数据变换操作仅构建逻辑计划，在触发 collect() / show() / write_*() 时统一优化执行，自动进行谓词下推、列裁剪等优化，减少不必要的 I/O 和计算开销。
丰富的数据格式支持。原生支持主流数据格式的读写，包括：
- 结构化数据：Parquet、CSV、JSON、SQL 数据库
- 数据湖格式：Delta Lake、Apache Iceberg、Apache Hudi、Lance
- AI 数据集：HuggingFace Hub 数据集
- 多媒体：视频帧序列、WARC 网页归档、MCAP 机器人传感器数据
多模态数据类型原生支持内置 Image、Video、Audio 等多媒体数据类型，支持直接在 DataFrame 列中存储和处理图像、音视频数据，无需手动序列化。
灵活的 UDF（用户自定义函数）系统 提供完整的 UDF 开发框架，支持声明 CPU、GPU、内存等资源需求，框架自动完成任务调度与资源分配。支持批处理模式、并发控制、进程/线程隔离，满足 CPU 密集与 GPU 推理等不同场景。
GPU 原生调度支持 UDF 可声明所需 GPU 资源（支持小数，如 num_gpus=0.5），框架与 Ray 协同完成 GPU 感知调度，天然适配深度学习推理、向量化等 GPU 密集型算子。
SQL 查询支持支持直接使用 SQL 语法对 DataFrame 进行查询（daft.sql()），降低数据处理门槛，兼容熟悉 SQL 的用户习惯。
多种存储后端统一接入通过统一的存储抽象层，支持本地文件系统、百度对象存储（BOS）、AWS S3、HTTP(S) 等多种存储后端，使用相同 API 访问不同存储，路径前缀自动路由。
DataFrame API 简洁直观提供类 Pandas 的 DataFrame 操作接口，支持 select、filter、groupby、join、sort、limit 等常用操作，以及窗口函数（Window），学习成本低。
数据湖 Catalog 集成支持与 Apache Iceberg、Apache Gravitino、Unity Catalog 等主流数据湖 Catalog 集成，实现数据治理、表版本管理与跨平台数据共享。

集成aihc-daft方式

你可以通过镜像或者 pip 包的方式，集成 aihc-daft。

镜像

Plain Text

1 ccr-registry.baidubce.com/aihc/aihc-daft-gpu:0.3.2-cu12.1-py3.11-ubuntu22.04

镜像已经预置aihc-Daft 以及多模态算子相关的依赖以及运行环境，比如 Cuda、Conda、Ray 等，推荐直接使用

pip包离线安装
- Aihc-daft 尚未在pypi 仓库发布，用户需要下载 aihc-daft 的离线安装包。点击下载
- 执行以下命令安装

                JSON
                
                pip install aihc_daft-0.3.2-cp310-abi3-manylinux_2_12_x86_64.whl

aihc-daft内置算子示例

这里以图片哈希计算处理器算子为例

test_image_hash.py脚本如下：

Plain Text

1from __future__ import annotations
2
3import os
4import daft
5from daft import col
6
7from daft.aihc.common.udf import aihc_udf
8from daft.aihc.functions.image.image_hash import ImageHash
9
10if __name__ == "__main__":
11    if os.getenv("DAFT_RUNNER", "native") == "ray":
12        import ray
13        ray.init(dashboard_host="0.0.0.0", ignore_reinit_error=True)
14        daft.set_runner_ray()
15    daft.set_execution_config(actor_udf_ready_timeout=6000, min_cpu_per_task=0)
16
17    samples = {
18        "image": [
19            "file:///local/sample_1.jpg",
20            "file:///mnt/pfs/sample_2.jpg",
21            "file:///mnt/bos/sample_3.jpg",
22        ]
23    }
24    
25    num_datasets = len(samples["image"]) 
26    ds = daft.from_pydict(samples).into_partitions(num_datasets) #强制分布式切分partitions
27    ds = ds.with_column(
28        "image_hash",
29        aihc_udf(
30            ImageHash,
31            construct_args={
32                "image_src_type": "image_url",
33                "method": "phash",
34            },
35            num_cpus=0.5,
36            batch_size=1,
37            concurrency=num_datasets,  # 多个数据集并发执行
38        )(col("image")),
39    )
40    ds.show()

分布式数据处理执行命令

Plain Text

1#使用DAFT_RUNNER=ray基于ray分布式执行
2DAFT_RUNNER=ray python test_image_hash.py 
3
4#单机执行
5python test_image_hash.py

aihc-daft 基础参数说明

aihc_udf 参数说明

参数	说明	默认值	示例
`operator`	算子类（必填）	—	`ImageHash`
`construct_args`	传给算子初始化的参数	`{}`	construct_args={ "image_src_type": "image_url", "method": "phash", }
`num_cpus`	每实例占用 CPU 核数	`None`（由调度器自动分配）	`2`
`num_gpus`	每实例占用 GPU 卡数	`None`（不使用 GPU）	`1` 或 `0.5`
`memory_bytes`	每实例内存上限（字节）	`None`（不限制）	`2 * 1024^3`（2GB）
`batch_size`	每次处理的数据条数	`None`（由框架自动决定）	`64`
`concurrency`	同时运行的实例数	`None`（由框架自动决定）	`8`
`use_process`	是否使用进程隔离（CPU 密集时建议开启）	`False`（使用线程）	`True`

数据读写方式

支持本地文件/挂载目录文件/BOS/HTTP等多种方式:

                Python
                
            

                    samples = {
        "image": [
            "file:///local/sample_1.jpg",      #本地数据
            "file:///mnt/pfs/sample_2.jpg",    #pfs挂载点内数据
            "file:///mnt/bos/sample_3.jpg",    #bos挂载点内数据
            "bos://bucket/path/sample_4.jpg"   #bos直接抓取数据
            "http://url/sample_5.jpg"          #http抓取数据
        ]
    }
            

若使用BOS直接抓取数据的方式, 需要在数据处理代码中注入BOS相关环境变量, 如下：

                Python
                
                os.environ["BOS_ENDPOINT"] = "http://bj.bcebos.com"     #endpoint
os.environ["BOS_ACCESS_KEY_ID"] = ""                    
os.environ["BOS_SECRET_ACCESS_KEY"] = ""
os.environ["BOS_REGION"] = "bj"

最佳实践

这里我们以具身数据格式转换为例，指导用户基于百舸平台的数据处理算子，通过开发机/分布式训练任务，实现 LerobotV2.1 数据集向 V3.0 版本的格式转换。准备工作

环境准备

这里我们可以使用开发机来开发调试代码，关于aihc-daft集成，你可以直接使用 aihc-daft 的镜像启动开发机，也可以使用自己的镜像启动，手动安装 aihc-daft 的包
数据准备
- 这里我们以huggingface中开源测试数据集lerobot/pusht/ 、dataset/lerobot/pusht2/和 lerobot/aloha_sim_insertion_human/ 为例
- 我们提供了打包的数据集，点击下载。原始数据集分布为:
  - dataset/lerobot/aloha_sim_insertion_human/
  - dataset/lerobot/pusht/
  - dataset/lerobot/pusht2/

算子开发

这里我们基于原始的TarUncompress算子，实现了递归目录下Tar 文件解压等能力。脚本如下：

                Python
                
            

                import json
import os
import tarfile
import daft
from daft import col

from daft.aihc.common.udf import aihc_udf
from daft.aihc.functions.process.tar_extractor_udf import TarUncompress
from daft.aihc.functions.process.tar_extractor_udf import discover_datasets
from daft.aihc.functions.process.tar_extractor_udf import create_tasks_from_datasets

TAR_EXTENSIONS = (".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tbz2", ".tar.xz", ".txz", ".tar.zst")


def is_tar_file(filepath: str) -> bool:
    lower = filepath.lower()
    if not any(lower.endswith(ext) for ext in TAR_EXTENSIONS):
        return False
    return tarfile.is_tarfile(filepath)


def find_tar_files(directory: str) -> list[str]:
    tar_files = []
    for root, _dirs, files in os.walk(directory):
        for f in files:
            full = os.path.join(root, f)
            if is_tar_file(full):
                tar_files.append(full)
    return tar_files


def safe_members(tf: tarfile.TarFile) -> list[tarfile.TarInfo]:
    return [m for m in tf.getmembers() if not m.name.startswith("/") and ".." not in m.name]


def extract_recursive(tar_path: str, output_dir: str) -> list[str]:
    all_extracted = []

    # 第一次解压
    with tarfile.open(tar_path) as tf:
        members = safe_members(tf)
        tf.extractall(path=output_dir, members=members)
        all_extracted.extend([os.path.join(output_dir, m.name) for m in members])

    # 持续扫描并解压新出现的 tar 文件
    pending = find_tar_files(output_dir)
    processed = set()

    while pending:
        current = pending.pop(0)
        if current in processed:
            continue
        processed.add(current)

        extract_dir = os.path.dirname(current)
        with tarfile.open(current) as tf:
            members = safe_members(tf)
            tf.extractall(path=extract_dir, members=members)
            all_extracted.extend([os.path.join(extract_dir, m.name) for m in members])

        # 删除已解压的内层 tar 包（如需保留，注释掉下面这行）
        # os.remove(current)

        new_tars = find_tar_files(output_dir)
        for t in new_tars:
            if t not in processed and t not in pending:
                pending.append(t)

    return all_extracted


class RecursiveTarUncompress(TarUncompress):
    """递归解压多层嵌套 tar 包的 UDF。"""

    def __call__(self, input_path, output_path):
        input_list = input_path.to_pylist()
        output_list = output_path.to_pylist()

        results = []
        for inp, outp in zip(input_list, output_list):
            os.makedirs(outp, exist_ok=True)
            all_files = extract_recursive(inp, outp)
            # 返回 JSON 字符串，与父类 TarUncompress 的 __return_column_type__(String) 一致
            results.append(json.dumps({
                "status": "success",
                "input": inp,
                "output": outp,
                "extracted_files": all_files,
                "extracted_count": len(all_files),
            }))
        return results


if __name__ == "__main__":
    if os.getenv("DAFT_RUNNER", "native") == "ray":
        import ray
        ray.init(dashboard_host="0.0.0.0", ignore_reinit_error=True)
        daft.set_runner_ray()
    daft.set_execution_config(actor_udf_ready_timeout=6000, min_cpu_per_task=0)

    base_path = "/mnt/pfs/xx"  # 【用户需替换】实际存放tar包的目录

    # 直接扫描 base_path 下的 tar 文件
    tar_files = find_tar_files(base_path)
    if not tar_files:
        raise ValueError(f"未在 {base_path} 下发现任何 tar 文件，请检查路径")

    # 去除所有 tar 后缀作为输出目录，如 test_dataset.tar.gz -> test_dataset
    def strip_tar_ext(path: str) -> str:
        while True:
            base, ext = os.path.splitext(path)
            if ext.lower() in (".gz", ".bz2", ".xz", ".zst", ".tar", ".tgz", ".tbz2", ".txz"):
                path = base
            else:
                break
        return path

    tasks = {
        "input_path": tar_files,
        "output_path": [strip_tar_ext(t) for t in tar_files],
    }

    num_tasks = len(tasks["input_path"])
    concurrency = max(num_tasks, 1)

    ds = daft.from_pydict(tasks)
    ds = ds.into_partitions(num_tasks)

    ds = ds.with_column(
        "result",
        aihc_udf(
            RecursiveTarUncompress,
            construct_args={},
            num_cpus=1,
            num_gpus=0,
            batch_size=1,
            concurrency=concurrency,
            use_process=True,
        )(col("input_path"), col("output_path")),
    )
    ds.show()
            

基于ConvertDatasetV21ToV30算子，将lerobotV2.1数据集格式转换为 lerobotV3.0数据集格式，脚本如下：

                Python
                
            

                import os
import daft
from daft import col

from daft.aihc.common.udf import aihc_udf
from daft.aihc.functions.embodied.convert_dataset_v21_to_v30_udf import ConvertDatasetV21ToV30

if __name__ == "__main__":
    if os.getenv("DAFT_RUNNER", "native") == "ray":
        import ray
        ray.init(dashboard_host="0.0.0.0", ignore_reinit_error=True)
        daft.set_runner_ray()
    daft.set_execution_config(actor_udf_ready_timeout=6000, min_cpu_per_task=0)

    tasks = {
        "input_repoid": [
            "lerobot/aloha_sim_insertion_human/", 
            "lerobot/pusht/",
            "lerobot/pusht2/"   
        ],
        "input_path": ["/mnt/pfs/xx/test_dataset/dataset/"] * 3,
        "output_path": ["/mnt/pfs/xx/lerobotv3"] * 3  # 【用户需替换】格式转换后的输出目录
    }
    num_datasets = len(tasks["input_repoid"])
    ds = daft.from_pydict(tasks).into_partitions(num_datasets)

    ds = ds.with_column(
        "convert_result",
        aihc_udf(
            ConvertDatasetV21ToV30,
            construct_args={
            },
            num_cpus=0.1,
            batch_size=1,
            concurrency=num_datasets,
            use_process=True
        )(col("input_repoid"), col("input_path"), col("output_path")),
    )
    ds.show()
            

整体处理流程:

运行pipiline.py

                Python
                
            

                import json
import os
import tarfile
import daft
from daft import col

from daft.aihc.common.udf import aihc_udf
from daft.aihc.functions.process.tar_extractor_udf import TarUncompress
from daft.aihc.functions.embodied.convert_dataset_v21_to_v30_udf import ConvertDatasetV21ToV30

# ====================== 直接复用 data_convert.py 全部代码 ======================
TAR_EXTENSIONS = (".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tbz2", ".tar.xz", ".txz", ".tar.zst")

def is_tar_file(filepath: str) -> bool:
    lower = filepath.lower()
    if not any(lower.endswith(ext) for ext in TAR_EXTENSIONS):
        return False
    return tarfile.is_tarfile(filepath)

def find_tar_files(directory: str) -> list[str]:
    tar_files = []
    for root, _dirs, files in os.walk(directory):
        for f in files:
            full = os.path.join(root, f)
            if is_tar_file(full):
                tar_files.append(full)
    return tar_files

def safe_members(tf: tarfile.TarFile) -> list[tarfile.TarInfo]:
    return [m for m in tf.getmembers() if not m.name.startswith("/") and ".." not in m.name]

def extract_recursive(tar_path: str, output_dir: str) -> list[str]:
    all_extracted = []
    with tarfile.open(tar_path) as tf:
        members = safe_members(tf)
        tf.extractall(path=output_dir, members=members)
        all_extracted.extend([os.path.join(output_dir, m.name) for m in members])

    pending = find_tar_files(output_dir)
    processed = set()

    while pending:
        current = pending.pop(0)
        if current in processed:
            continue
        processed.add(current)

        extract_dir = os.path.dirname(current)
        with tarfile.open(current) as tf:
            members = safe_members(tf)
            tf.extractall(path=extract_dir, members=members)
            all_extracted.extend([os.path.join(extract_dir, m.name) for m in members])

        new_tars = find_tar_files(output_dir)
        for t in new_tars:
            if t not in processed and t not in pending:
                pending.append(t)

    return all_extracted

class RecursiveTarUncompress(TarUncompress):
    """递归解压多层嵌套 tar 包的 UDF。"""
    def __call__(self, input_path, output_path):
        input_list = input_path.to_pylist()
        output_list = output_path.to_pylist()
        results = []
        for inp, outp in zip(input_list, output_list):
            os.makedirs(outp, exist_ok=True)
            all_files = extract_recursive(inp, outp)
            results.append(json.dumps({
                "status": "success",
                "input": inp,
                "output": outp,
                "extracted_files": all_files,
                "extracted_count": len(all_files),
            }))
        return results

# ====================== 主 pipeline 工作流 ======================
if __name__ == "__main__":
    # 统一环境初始化
    if os.getenv("DAFT_RUNNER", "native") == "ray":
        import ray
        ray.init(dashboard_host="0.0.0.0", ignore_reinit_error=True)
        daft.set_runner_ray()
    daft.set_execution_config(actor_udf_ready_timeout=6000, min_cpu_per_task=0)

    base_path = "/mnt/pfs/xx"
    convert_output_root = "/mnt/pfs/xx/lerobotv3"

    # ====================== 步骤1：执行解压（原 data_convert.py） ======================
    print("=== 步骤1：开始递归解压 tar 文件 ===")
    tar_files = find_tar_files(base_path)
    if not tar_files:
        raise ValueError(f"未在 {base_path} 下发现任何 tar 文件")

    def strip_tar_ext(path: str) -> str:
        while True:
            base, ext = os.path.splitext(path)
            if ext.lower() in (".gz", ".bz2", ".xz", ".zst", ".tar", ".tgz", ".tbz2", ".txz"):
                path = base
            else:
                break
        return path

    tasks_extract = {
        "input_path": tar_files,
        "output_path": [strip_tar_ext(t) for t in tar_files],
    }
    num_tasks = len(tasks_extract["input_path"])
    concurrency = max(num_tasks, 1)

    ds = daft.from_pydict(tasks_extract)
    ds = ds.into_partitions(num_tasks)

    ds = ds.with_column(
        "result",
        aihc_udf(
            RecursiveTarUncompress,
            construct_args={},
            num_cpus=1,
            num_gpus=0,
            batch_size=1,
            concurrency=concurrency,
            use_process=True,
        )(col("input_path"), col("output_path")),
    )
    df_extract = ds.collect()
    print("=== 解压完成 ===")

    # ====================== 步骤2：执行 v21 → v30 转换（原 lerobotv21-30.py） ======================
    print("=== 步骤2：开始格式转换 ===")
    tasks_convert = {
        "input_repoid": [
            "lerobot/aloha_sim_insertion_human/",
            "lerobot/pusht/",
            "lerobot/pusht2/"
        ],
        "input_path": ["/mnt/pfs/xx/test_dataset/dataset/"] * 3,
        "output_path": [convert_output_root] * 3
    }
    num_datasets = len(tasks_convert["input_repoid"])

    ds_convert = daft.from_pydict(tasks_convert).into_partitions(num_datasets)

    ds_convert = ds_convert.with_column(
        "convert_result",
        aihc_udf(
            ConvertDatasetV21ToV30,
            construct_args={},
            num_cpus=0.1,
            batch_size=1,
            concurrency=num_datasets,
            use_process=True
        )(col("input_repoid"), col("input_path"), col("output_path")),
    )
    ds_convert.show()
    print("=== 全部 pipeline 执行完成 ===")
            

分布式数据处理

在分布式训练模块中，基于上述开发的算子代码，使用 Ray 计算引擎进行分布式处理数据。

也可以直接使用开发机进行单机的数据处理

Rayjob的提交可参考快速提交Ray任务，关键参数如下：

镜像地址：使用百舸预置镜像，选择aihc-daft预置镜像
执行命令：DAFT_RUNNER=ray python /mnt/pfs/xx/pipeline.py，其中pipeline.py是上面开发的算子代码。
计算框架：选择 Ray
计算资源：可设置多个 worker 实例并行执行。在任务执行时，Daft + Ray 会自动调度、自动负载均衡、用满集群资源。
存储挂载：将源数据所在存储实例挂载到容器内

提交任务即可进行数据处理，可通过 submitter 节点的日志查询数据处理的进度

评价此篇文章

有帮助没帮助

算子列表

数据集管理

百度智能云

百度百舸 · AI计算平台

百度百舸 · AI计算平台

AIHC-Daft算子开发使用指南

Daft 核心特性

集成aihc-daft方式

aihc-daft内置算子示例

aihc-daft 基础参数说明

aihc_udf 参数说明

数据读写方式

最佳实践

环境准备

算子开发

分布式数据处理