引言：具身智能时代的数据工程挑战

大模型在多模态领域掀起技术浪潮之后，具身智能（Embodied AI）成为业界公认的下一个前沿高地。与传统AI依赖互联网上静态图文不同，具身智能的训练高度依赖来自真实物理世界的多模态感知数据——第一人称视角（Egocentric）视频、深度传感器数据、力反馈信号，缺一不可。

举个具体案例，要训练一个能理解厨房操作的机器人，通常需要采集数千小时的人类佩戴摄像头拍摄的操作视频。解码之后，这些数据会产生数亿张图像帧。但问题在于，原始数据里充斥着大量“噪音”：静止画面、模糊镜头、无效空帧。核心难题在于：如何从这海量的冗余信息中，高效地提取出高质量样本？这已经成为制约具身智能发展的关键瓶颈。

针对这一行业痛点，阿里云 EMR Serverless Spark 引入了 Ray 分布式计算框架，并深度集成了一款专为多模态 AI 工作负载设计的高性能数据引擎——Daft。接下来，我们将深入解析如何利用这套组合拳，构建一条从视频抽帧、清洗到标注的全自动化流水线。

一、 Daft：为 AI 多模态而生的高性能数据引擎

1.1 为什么需要 Daft？

Daft 的定位非常明确——它是为 AI 时代的非结构化数据量身打造的分布式 DataFrame 引擎。在大模型驱动的智能应用中，数据早已超越了传统意义上的结构化表格。视频、图像、音频、Embedding 这些非结构化数据，才是大模型真正的“燃料”。传统数据处理引擎面对这些复杂类型时往往力不从心，而 Daft 的诞生，正是为了解决这个问题：

• 多模态原生支持：Daft 从底层类型系统就把 Image、Video、Audio、Tensor 视为“一等公民”，天然理解它们的物理布局，并能进行针对性的优化。
• 极简开发体验：开发者只需一行表达式，就能在分布式集群上完成视频解码、帧提取、图像变换，底层的分片调度完全不用操心。
• “数据处理即 AI 推理”：通过 UDF 机制，Daft 允许将通义千问等大模型的推理能力直接嵌入 DataFrame 流水线。图像理解、视频描述生成这类 AI 任务可以与数据清洗无缝衔接，原本割裂的多套系统，最终收敛为一条统一的 Python Pipeline。

1.2 架构优势

• Rust 核心 + Arrow 列式内存：Daft 以 Python API 暴露数据处理能力，底层由 Rust 执行引擎支撑，并利用 Apache Arrow 生态的列式内存表示，大幅降低了序列化与数据交换的开销。
• 面向 AI/多模态负载的执行优化：Daft 采用惰性求值，执行前会进行规则优化和代价优化，比如谓词/投影下推、Join 重排等。而对于 UDF、模型推理、下载、解码等高开销环节，Daft 通过 Source、UDF 和 AI Functions 等机制进行任务化执行，并支持批处理、并发控制和自适应背压。
• CPU 与 GPU 资源调度能力：常规数据处理由本地执行引擎或分布式 runner 执行；模型推理这类 GPU 任务，则可以通过 AI Functions 或 UDF 声明 GPU 资源、batch size 与并发度，在吞吐、显存和延迟之间做精细调优。
• 从单机到集群扩展：同一套 DataFrame/SQL API 可以在本地 native runner 上运行，也能通过 Ray 扩展到分布式集群，适合从开发环境平滑迁移到大规模处理场景。

二、 EMR Serverless 集成 Daft 引擎的价值

阿里云 EMR Serverless Spark 在其 Serverless 架构中深度集成了 Ray 分布式计算框架，并内置了 Daft 引擎，提供了一套开箱即用的多模态数据处理基础设施。

1. 极致的弹性与免运维 依托 EMR Serverless 架构，用户无需关心底层集群的搭建、扩缩容和节点维护。只需聚焦于业务逻辑开发，按需使用、按量计费，真正实现了多模态数据处理基础设施的“零运维”，大幅降低了 TCO。

2. 内置多模态算子，AI 与数据无缝融合EMR Serverless Spark 在 Daft 引擎中内置了丰富的多模态 AI 函数，支持图片、视频、音频、文本等数据的直接处理。同时，与阿里云百炼平台上的大语言模型（如 Qwen 系列）无缝集成，让数据清洗、特征提取与 AI 标注可以在同一流水线中高效完成，加速 AI 应用落地。

3. 统一的数据湖体验，打通存储孤岛EMR Serverless Spark 提供了与阿里云 OSS、OSS-HDFS 等对象存储的深度集成，并支持无缝挂载 NAS 和 CPFS 高性能文件系统，满足模型训练与推理阶段对海量非结构化数据的高吞吐读写需求，实现结构化与非结构化数据的统一访问与管理。

4. 稳定性提升Daft 基于 Ray 的架构天然支持异构资源，EMR Serverless Daft 在异构节点的任务调度时能做到细粒度控制，避免关键节点（如 Header）负载过高，解决部分算子因为依赖配置而非实际资源引起的作业 Hang 住问题。同时实现对内存的监管，针对核心算子基于运行时内存使用情况支持 Spill 机制，有效解决大数据量规模下算子运行的稳定性问题。

5. 可观测性提升Dashboard 以及细粒度的 Metrics 是分析和优化作业的核心参考。EMR Serverless Daft 提供了全托管的 Daft Dashboard，保证云上作业信息随时可查；同时基于在 Spark 领域的丰富经验，拓展了多项算子和 UDF（包括 AI Function）等运行时指标信息，全面提升作业的可观测性。

6. 融入 OpenLake 生态，释放数据要素价值作为阿里云 OpenLake 多模态解决方案的关键组件，EMR Serverless Daft 全面支持 Data Lake Formation（DLF）与 Apache Paimon。依托 DLF 对湖上元数据的统一管理，借助 Paimon 的 Blob、Vector 等多模态原生能力，进一步发挥 Daft 引擎在“结构化+多模态”混合处理场景中的独特优势，助力企业构建统一的智能数据底座。

三、 具身智能实践：从原始视频到高质量训练数据的全流程流水线

3.1 场景背景

训练具身智能模型理解和复现人类日常操作动作时，研发人员通常会让真实人类佩戴头戴式摄像头，在厨房等真实生活场景中执行烹饪、切菜、翻炒等操作，采集大量第一人称视角（Egocentric）的原始视频。

如下图所示，这类视频记录了人类在厨房中翻炒食材的完整过程，是训练 VLA（视觉-语言-动作）模型的宝贵数据来源。

这些视频中包含大量无价值的冗余帧——例如操作间隙的静止画面、模糊镜头，或者镜头朝向地面/天花板时拍摄到的无效内容。我们需要一个高效的自动化流水线，从海量视频中抽取关键帧，并利用多模态大模型对图片进行智能筛选，最终只保留包含有效烹饪操作内容（例如"锅具"或"食材处理"）的高质量图片帧，用于后续的模型训练。

3.2 流水线架构

整个流程分为五个核心步骤，实现了从原始视频到高质量数据集的端到端自动化转化：

步骤 1：分布式视频读取与高效抽帧

利用 Daft 的流式优化能力，直接在 OSS 上进行分布式读取和解码，避免本地下载开销。

import daft
from daft import col

# 流式读取 OSS 视频，仅提取关键帧，每2秒采样一次
df = daft.read_video_frames(
    path="oss:///cooking_videos/*.mp4",
    image_height=480,
    image_width=640,
    is_key_frame=True,   # 只取关键帧，减少冗余
    sample_interval_seconds=2.0, # 采样频率
    max_frames_per_video=100,    # 限制单视频最大帧数
    io_config=io_config,
)

步骤 2：图像编码与批量写入 OSS

抽取的原始视频帧是未经压缩的张量，直接存储成本极高。Daft 提供了 encode_image 函数，可将帧数据即时编码为 JPEG 格式。随后，通过 FrameUploader 批量 UDF 将图片写入 OSS，并获取每张图片的 OSS 路径：

from daft.functions import encode_image
from daft.emr.functions import emr_udf

# 1. 编码为 JPEG
df = df.with_column("jpeg_bytes", encode_image(col("data"), "JPEG"))

# 2. 批量上传至 OSS (并发控制最大化 I/O 吞吐)
df = df.with_column(
    "oss_path",
    emr_udf(
FrameUploader,
construct_args={"output_base": output_base},
concurrency=4,
batch_size=32,
    )(col("path"), col("frame_index"), col("jpeg_bytes")),
)

FrameUploader 内部使用 Daft 内置的 io_put 接口写入 OSS，无需引入额外的 oss2 SDK，并且通过 concurrency 参数控制并发写入数量，最大化网络 I/O 吞吐量。

步骤 3：集成多模态大模型进行智能筛选

这是最关键的一步。利用 ai_query 函数，让 Qwen-VL 大模型直接通过签名 URL 读取 OSS 图片进行推理，无需将图片数据拉回计算节点再转发给模型网关，极大地节省了带宽并降低了延迟：

from daft.emr.functions import ai_query
from daft import lit

# 调用 Qwen-VL 进行视觉判断
df = df.with_column(
    "llm_result",
    ai_query(
prompt=lit("判断这张图片是否包含有效的烹饪操作画面（如锅具、食材处理、手部操作等）。如果包含请只回复KEEP，不包含请只回复DROP，不要输出其他内容"),
data=col("oss_path"),# 传入 OSS 路径，模型服务端直接拉取
model="qwen-vl-plus",
concurrency=32,    # 32并发，充分利用模型吞吐
batch_size=32,
options={"enable_thinking": False},
    ),
)

通过配置 max_concurrency=32，流水线将启动 32 个并发的 UDF 实例，充分利用模型服务的吞吐能力，大幅缩短整体处理时间。

步骤 4：结果解析与冗余数据自动清理

解析 LLM 返回的结果，保留标记为 "KEEP" 的图片，物理删除标记为 "DROP" 的无效数据。

from daft.functions import get as struct_get

df = df.with_column("llm_content", struct_get(col("llm_result"), "content"))
df = df.with_column("keep", col("llm_content").upper().contains("KEEP"))

# 物化结果，统计保留与删除数量
df = df.collect()
result_dict = df.to_pydict()
total = len(result_dict["keep"])
kept = sum(1 for k in result_dict["keep"] if k)
dropped = total - kept

# 删除 LLM 判定为 DROP 的帧
if dropped > 0:
    df_drop = df.where(col("keep") != True).select("oss_path")
    df_drop = df_drop.with_column(
"deleted",
emr_udf(OSSDeleter, concurrency=4, batch_size=32)(col("oss_path")),
    )
    df_drop.collect()

3.3 流程总结

通过这四个步骤，我们构建了一个端到端、全自动的具身智能数据处理流水线。整个流程的数据流向如下表所示：

原始的第一人称视角烹饪视频输入，高质量、经过大模型标注和清洗的训练图片集输出。整个过程无需人工干预，且能够随着 Ray 集群的弹性伸缩轻松应对 PB 级的数据规模。

四、 EMR Serverless Ray & Daft 产品架构与生态能力

4.1 灵活的运行模式

Daft 作为新一代多模态 DataFrame 框架，向下无缝对接弹性计算资源，向上提供统一的数据处理 API。在底座层面，Ray 引擎实现了 CPU 与 GPU 队列组的混合调度，并全面接入 Kerberos 与 Ranger，为企业级应用提供坚实的安全隔离保障。

针对不同阶段的业务需求，平台提供了三种灵活的运行模式：

1. Local Mode（本地模式）：该模式在单 Pod 容器内直接执行，零启动开销。其按需计费、用完即释放的特性，是轻量级代码调试与单机验证的理想选择。
2. On Ray - Ephemeral（临时集群模式）：专为大规模 CPU&GPU 批处理设计。系统在任务提交时自动拉起临时 Ray 集群，任务完成后自动销毁，用户仅需指定计算规格，实现基础设施的完全免运维和任务之间的完全隔离。
3. On Ray - Persistent（常驻集群模式）：主要面向 CPU&GPU 多模态数据处理，深度学习推理及生产环境。该模式支持将任务路由至常驻 Ray 集群，实现计算资源的长期复用，并通过统一的 Dashboard 提供完善的作业监控。

4.2 100+ 开箱即用算子

在业务落地过程中，非结构化数据的预处理往往是工程链路上最耗时的环节。为此，EMR Serverless Spark 平台在 Daft 框架内深度沉淀了 100 余个开箱即用的多模态处理算子，全面覆盖视频、文本/文档、图片及音频四大核心场景。

在视频与图片处理方面，算子库涵盖了从基础的切片、抽帧、格式转换，到高阶的视频理解、人脸模糊及 OCR 识别；在文本与文档领域，不仅提供标准化的切分与去重功能，更针对 RAG（检索增强生成）场景内置了 PDF 智能解析与向量化提取能力；同时，音频处理也实现了从格式互转到多语种识别的全链路覆盖。

企业无需投入高昂的算法研发成本，也无需自行拼凑开源组件，即可快速构建高可用、企业级的多模态数据清洗与转换流水线，显著缩短 AI 业务的上线周期。

4.3 API 封装释放大模型生产力

为了进一步降低分布式数据处理的门槛，Daft 算子在设计上被划分为两类：负责传统数据过滤与聚合的基础算子，以及深度集成百炼等大模型服务的 AI 算子。

以“视觉理解（VisionUnderstanding）”算子为例，在传统的开发模式下，开发者需要自行处理模型加载、网络请求及分布式并发控制。而在 EMR Serverless 平台中，用户仅需通过数行极简代码，即可从 OSS 读取多模态数据，并调用 AI 算子完成复杂的推理任务。

这种设计的价值在于：

• 开箱即用：免除繁琐的模型部署。
• 极简 API：数行代码完成复杂推理。
• 分布式透明化：底层由 Ray 自动接管并发调度与批处理加速。

五、 适用场景

这套技术栈适合以下几类典型场景：

具身智能行为数据集构建

从佩戴摄像头采集的第一人称视角原始视频中，自动完成抽帧、质量过滤、目标识别标注，构建 VLA（视觉-语言-动作）模型训练所需的高质量图文对数据集。

视频内容理解与特征提取

对监控视频、工业检测视频等进行批量帧级分析，提取关键事件帧，用于下游的模型训练或业务分析。

大规模多模态数据清洗与标注

对互联网爬取的图文、视频数据进行去重、质量评估、内容分类，为多模态大模型的预训练提供高质量数据源。

结语

阿里云 EMR Serverless Spark 引入 Ray 分布式计算框架与 Daft 高性能数据引擎，为用户提供了一套开箱即用、免运维且极致高效的多模态数据处理基础设施。正如在具身智能视频抽帧与 LLM 筛选实践中看到的，Daft on Ray 打破了传统批处理引擎在内存管理和异构资源调度上的瓶颈，让数据工程师和 AI 研究员能够以极简的代码，构建出稳定、高效的大规模多模态数据流水线，大幅加速具身智能数据飞轮的迭代速度，为下一代具身智能模型的研发提供坚实的数据基础。