Daft 图像分类基准测试：80 万张图片的分布式 GPU 推理流水线，以及与 Ray Data、Spark 的实测对比-酒店常州论坛

Daft 图像分类基准测试：80 万张图片的分布式 GPU 推理流水线，以及与 Ray Data、Spark 的实测对比

【免费下载链接】DaftHigh-performance data engine for AI and multimodal workloads. Process images, audio, video, and structured data at any scale项目地址: https://gitcode.com/GitHub_Trending/da/Daft

本文基于 Daft 仓库中的图像分类基准测试文档，完整解析这一多模态 AI 基准测试的工作负载设计、三引擎（Daft / Ray Data / Spark）实测结果，以及 Daft 实现中的关键源码机制：@daft.clsGPU 模型封装、download/decode_image原生算子、批量推理方法与集群资源配置。读完本篇，你可以理解如何在 Ray 集群上跑通一条完整的"读 Parquet → 下载图片 → 解码 → 预处理 → ResNet18 推理 → 写 Parquet"的多模态推理流水线，并掌握其中的关键参数与调优点。

基准测试工作负载设计

图像分类基准测试的目标是模拟真实的大规模 AI 推理场景。其官方定义如下（引自 README）：

数据规模：共803,580 行数据，由 80,358 张唯一图片各重复 10 次构成，用于放大推理负载、验证分布式扩展性；
模型：TorchVision 的 ResNet18（ImageNet 预训练权重）；
处理流程：下载图片 → 应用预处理变换 → 运行推理，预测 ImageNet 类别标签，任务分布在多个 GPU 节点上；
输入数据集：ImageNet 基准数据集，S3 Parquet 格式（位于s3://daft-oss-public-datasets/imagenet/benchmark，每行含image_url字段）；
输出格式：Parquet，包含图片 URL 与预测标签两列；
集群规格：8 个 GPU worker 节点，AWSg6.xlarge实例（每节点 1 块 L4 GPU）；
基准测试日期：2024 年 9 月 22 日；
框架版本：Daft 0.6.2、Ray Data 2.49.2、AWS EMR Spark 7.10.0。

该基准是仓库 AI 基准测试套件的四个工作负载之一，同套件还包括音频转写、文档嵌入与视频目标检测，共同覆盖多模态数据处理的不同侧面。

性能结果

同一数据集、同一集群规格下，三引擎的实测运行时间如下（来自 README）：

引擎	运行时间
Daft	4m 23s
Ray Data	23m 30s
Spark	45m 7s

AI 基准汇总页中给出了同一数据规模（803,580 张图）下的一致记录：Daft 4m 23s、Ray Data 23m 30s、Spark 45m 7s。这一结果展示了在多模态"下载 + 解码 + 模型推理"混合负载下各引擎的端到端差异——其中图片下载与解码这类 CPU/IO 密集步骤，以及 GPU 推理步骤之间的流水化程度，是影响总时长的关键。

Daft 实现解析：daft_main.py

核心实现在 daft_main.py，整条流水线约 80 行代码，可拆分为五个环节。

1. 集群就绪等待与运行器配置

NUM_GPU_NODES = 8 daft.set_runner_ray() # Wait for Ray cluster to be ready @ray.remote def warmup(): pass ray.get([warmup.remote() for _ in range(64)])

脚本先通过daft.set_runner_ray()指定 Ray 作为执行后端；随后提交 64 个空的@ray.remote任务（8 节点 × 8），等待 Ray 自动扩容把 8 个 worker 全部拉起后再开始正式计时，避免把集群冷启动时间计入基准结果。

2. GPU 模型类：@daft.cls 封装 ResNet18

weights = ResNet18_Weights.DEFAULT transform = transforms.Compose([transforms.ToTensor(), weights.transforms()]) @daft.cls( max_concurrency=NUM_GPU_NODES, gpus=1, ) class ResNetModel: def __init__(self): self.weights = weights self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") self.model = resnet18(weights=weights).to(self.device) self.model.eval() @daft.method.batch( return_dtype=daft.DataType.string(), batch_size=BATCH_SIZE, ) def __call__(self, images): if len(images) == 0: return [] torch_batch = torch.from_numpy(np.array(images.to_pylist())).to(self.device) with torch.inference_mode(): prediction = self.model(torch_batch) predicted_classes = prediction.argmax(dim=1).detach().cpu() predicted_labels = [self.weights.meta["categories"][i] for i in predicted_classes] return predicted_labels

结合 Daft 类 UDF 的源码定义，这里涉及两个关键机制：

@daft.cls模型复用：该装饰器把普通 Python 类转为 Daft 用户自定义类，每个实例的__init__只在查询执行时被惰性调用一次，之后同一实例复用于多批数据。对于"加载模型权重并绑定 CUDA 设备"这类昂贵初始化，按 8 个 actor 各加载一次而非按行加载，是 GPU 推理 UDF 的标准写法。参数方面：
- gpus=1：每个实例申请 1 块 GPU（源码支持 0~1 之间的小数，用于多个小模型共享一张 GPU，但不支持大于 1 的小数）；
- max_concurrency=NUM_GPU_NODES：同步方法下控制 actor 池规模，即最多 8 个并发推理实例，正好对应 8 个 GPU 节点；
- 另有cpus、use_process、max_retries、on_error（"raise"/"log"/"ignore"）、ray_options等可选参数可进一步调优。
@daft.method.batch批量推理：将方法声明为按批调用，batch_size=100与文件顶部BATCH_SIZE = 100一致，即每次向 GPU 送 100 张图片；return_dtype=daft.DataType.string()显式声明输出列类型为字符串（标签名）。torch.inference_mode()与argmax取预测类别，再通过weights.meta["categories"]把类别索引映射为 ImageNet 标签名。

3. 端到端流水线表达式

daft.set_planning_config( default_io_config=daft.io.IOConfig(s3=daft.io.S3Config.from_env().replace(requester_pays=True)) ) df = daft.read_parquet(INPUT_PATH) df = df.with_column( "decoded_image", df["image_url"].download().decode_image(mode=daft.ImageMode.RGB), ) df = df.with_column( "norm_image", df["decoded_image"].apply( func=lambda image: transform(image), return_dtype=daft.DataType.tensor(dtype=daft.DataType.float32(), shape=IMAGE_DIM), ), ) df = df.with_column("label", ResNetModel()(col("norm_image"))) df = df.select("image_url", "label") df.write_parquet(OUTPUT_PATH)

这条链路的每个环节对应一个原生或 UDF 算子：

daft.read_parquet(INPUT_PATH)：读取 S3 上的 Parquet。注意IOConfig中requester_pays=True——输入桶属于 S3 请求者付费桶，必须显式开启计费模式才能访问；
image_url.download()：把 URL 字符串列逐行下载为字节列。从表达式定义看，默认max_connections=32（每分区并发连接数）、on_error="raise"，下载逻辑本身是引擎级并发 IO，而非逐行 Python 循环；
.decode_image(mode=daft.ImageMode.RGB)：将字节解码为统一 RGB 模式的图像列，同样是 Rust 引擎内的原生算子（见 decode_image 表达式），支持on_error="raise"/"null"错误策略；
apply预处理：用 TorchVision 的transforms.Compose([ToTensor(), weights.transforms()])把图片转为(3, 224, 224)的 float32 张量列，return_dtype=daft.DataType.tensor(dtype=float32, shape=IMAGE_DIM)显式声明张量 schema，让引擎可以在后续 stage 间正确传递该列；
ResNetModel()(col("norm_image"))：调用上一步定义的 Daft 类 UDF 做 GPU 推理，产出label列；
select+write_parquet：只保留image_url与label两列写出 S3，对应 README 中"Parquet with image URLs and predicted labels"的输出格式。

整条流水线是惰性声明的：从读表到写表没有显式的.collect()，write_parquet触发一次性执行，下载、解码、预处理、推理各阶段在引擎内部流水化重叠执行。

4. 计时方式

脚本用time.time()在read_parquet之前开始计时、在write_parquet之后结束，打印总秒数。而在 CI 场景中，仓库提供了统一入口 run_ai_benchmark.py：它以 Ray Job Submission 方式提交DAFT_RUNNER=ray DAFT_PROGRESS_BAR=0 python daft_main.py作为 entrypoint，先做一次 warmup 运行，再正式运行 2 次取平均，最后把结果连同 Daft 版本等元数据上传记录。

Ray Data 对照实现

ray_data_main.py 用相同的 ResNet18 模型、相同输入输出路径与BATCH_SIZE = 100构建了对照流水线：

paths = ray.data.read_parquet(INPUT_PATH).take_all() paths = [row["image_url"] for row in paths] ds = ( ray.data.read_images(paths, include_paths=True, ignore_missing_paths=True) .map(fn=transform_image) .map_batches(fn=ResNetActor, batch_size=BATCH_SIZE, num_gpus=1.0, concurrency=NUM_GPU_NODES) .select_columns(["path", "label"]) ) ds.write_parquet(OUTPUT_PATH)

两个实现的可对照差异点：

图片获取：Ray Data 侧先用take_all()把全部 80 万条 URL 一次性拉到驱动进程，再交给read_images下载；Daft 侧则保持 URL 列在分布式数据内，用引擎级download()算子在各分区并发下载；
逐行 vs 批式算子：Ray Data 用.map（逐行函数transform_image，内部Image.fromarray(row["image"]).convert("RGB")）做预处理，.map_batches(ResNetActor, num_gpus=1.0, concurrency=8)做推理；Daft 侧预处理走apply+ 张量列，推理走@daft.cls+@daft.method.batch；
GPU 资源表达：num_gpus=1.0/concurrency=8与 Daft 的gpus=1/max_concurrency=8语义对应，两边的 actor 池规模一致。

Spark 对照实现

spark.ipynb 基于 AWS EMR Spark 7.10.0 完成同一任务，其要点：

通过%%configure设置spark.sql.execution.arrow.maxRecordsPerBatch = 100，将 Arrow 批大小对齐到 100，与前两个实现的BATCH_SIZE = 100保持一致；
用模块级_model_cache字典在 executor 进程内缓存resnet18模型、权重与设备（并把TORCH_HOME/XDG_CACHE_HOME指到/tmp），避免重复加载权重；
图片解码与预处理封装为@pandas_udf（decode_and_preprocess_image_udf），逐条Image.open字节流、convert("RGB")后过transform，以ArrayType(FloatType())返回；推理同样以 pandas UDF 形式完成。

集群与依赖配置

Ray 集群配置

cluster.yaml 定义了 CI 使用的 Ray 集群（AWSus-west-2）：

节点类型：head 节点不占 CPU/GPU 资源（resources: {"CPU": 0, "GPU": 0}）；worker 固定min_workers: 8/max_workers: 8，保证扩容后精确为 8 个 GPU worker，避免自动缩放在测试期间扰动结果；
实例规格：head 与 worker 均为g6.xlarge（每节点 1 块 L4 GPU），PyTorch AMI（ImageId: ami-0976479b866d22613），100GB gp3 加密云盘；
安全与 IAM：统一安全组ray-autoscaler-c1、IAM 角色ray-autoscaler-v1，SSH 密钥ci-github-actions-ray-cluster-key；
环境准备：setup_commands中把 Ray 的 tmp 与 object spilling 目录指到/opt/ray（防止对象溢出打爆根盘），并安装固定版本依赖：ray[default]==2.49.2、numpy==1.26.4、torchvision==0.22.0+cu128、pillow==11.3.0，Daft 则以pip install daft --pre --extra-index-url ${DAFT_INDEX_URL}从预发布索引安装。

依赖锁定

pyproject.toml 锁定本基准的运行时环境：Python==3.12，daft==0.6.2，ray[default]==2.49.2，与 README 中的"Framework Versions"一一对应，保证结果可复现。

结论：这个基准说明了什么

该基准测量的是多模态推理的端到端能力，而非单纯的模型吞吐：图片下载（网络 IO）、解码（CPU）、预处理（CPU/张量转换）、GPU 推理（CUDA）与结果落盘（S3 写）串在一条查询里。Daft 在此场景下的 4m 23s（相对 Ray Data 23m 30s 与 Spark 45m 7s）的优势，从源码结构看主要来自三方面：下载/解码作为引擎原生算子与 GPU 推理 stage 流水化重叠执行（而非 Ray Data 中先take_all再read_images的两段式路径）；@daft.cls+@daft.method.batch让模型按 actor 池粒度加载一次、按 100 条/批送 GPU；以及显式张量列 schema（DataType.tensor）使预处理输出能以紧凑的 Arrow 张量列在 stage 间传递。若你在自己的仓库中复现类似负载，可直接以 daft_main.py 为模板，替换INPUT_PATH/OUTPUT_PATH与模型类，并按 cluster.yaml 的思路固定 GPU worker 数量以保证结果稳定。

需要注意的适用前提：输入桶为 S3 请求者付费桶（requester_pays=True），模型为 TorchVision 官方 ImageNet 权重，且结果基于 2024 年 9 月 22 日的 Daft 0.6.2 / Ray Data 2.49.2 / EMR Spark 7.10.0 版本组合，跨版本对比时应先对齐框架版本与集群规格。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析