阿里云 EMR Serverless Spark 重磅推出全托管 Ray，在同一工作空间内支持 Spark 和 Ray，并以统一湖仓存储、统一 CPU&GPU 资源池、统一安全和运维体系，承载从结构化数据处理到 AI 训练数据预处理、后训练、推理与服务的完整链路。面向具身智能、自动驾驶等多模态场景，企业由此可以在一套云原生平台上完成数据治理、样本加工、模型开发和规模化生产。

数据与 AI 深度融合，需要贯通全链路的计算底座

生成式 AI 正在改变企业的数据处理对象。传统数据工程主要面向结构化表、日志和文本；具身智能、自动驾驶等新一代 AI 系统工程则持续产生第一视角视频、连续图像、语音、传感器记录和控制轨迹。数据规模扩大之外，数据形态、算子类型和资源需求也在发生变化：一条生产流水线中，既有大规模 SQL、Join、去重和统计，也有视频解码、图像理解、向量化、样本过滤、模型推理等 Python 原生任务，并同时消耗 CPU、GPU、网络和存储带宽。

具身智能中的 egocentric 数据是典型代表。机器人或可穿戴设备以第一视角记录人与环境的持续交互，单个样本往往同时包含视频片段、语音、文本指令、动作序列和时间戳。自动驾驶同样需要围绕海量路测数据完成切帧、场景识别、长尾目标发现、标注、去重、特征生成和回灌。当前大量视觉、语音、强化学习和大模型工具均围绕 Python 生态构建，这些任务需要可横向扩展的 Python 分布式运行环境；与此同时，样本目录管理、批量 ETL、数据质量、湖表写入和跨任务治理，仍然需要成熟的大数据处理能力。

因此，阿里云 EMR Serverless Spark 引入全托管 Ray，Spark 负责规模化数据处理、SQL 分析、湖仓读写；Ray 负责 Python 原生分布式计算、多模态数据处理、训练推理和模型服务。二者面向同一业务流程协同工作，共同构成 Data + AI 一体化计算底座。为用户提供面向数据与 AI 工作负载的全托管 Lakehouse 引擎。

架构演进：极致全模态内核优化，统一数据和算力底座

Spark Fusion2.0：全球第一的结构化处理引擎

EMR Serverless Spark 的结构化执行引擎 Fusion2.0在 TPC-DS 100T 官方 Benchmark 取得世界第一的成绩，相比前榜首性能提升100%，性价比提升500%。

除了标准Benchmark，依托每日海量规模的生产实践，Fusion2.0针对复杂的生产Workload也做了大量QO和QE联合优化，例如针对吞吐和延迟的IO优化，针对多场景数据倾斜的优化，针对半结构化数据的解析优化，针对超大规模Shuffle的优化，针对磁盘溢写的优化，基于历史的优化等。得益于这些优化，海量真实生产作业也获得数倍性能提升。

EMR Ray: 稳定高性能的多模态处理框架

Ray是全栈AI计算框架，包含Ray Core和Ray AI Libraries，涵盖资源管理、高性能存储、计算原语、多模态数据处理、AI训练、AI推理服务等重要功能。

EMR Ray Core做了诸多优化提升稳定性，包括Ray Head高可用，多级节点容错，基于磁盘水位的自动扩缩容，GCS外挂高可用Redis，坏卡自动检测隔离等。

EMR Ray的多模态处理能力主要来自Ray Data，Daft on Ray，Datajuicer on Ray。多模态处理和结构化处理并非割裂，而是侧重点不同，结构化处理侧重关系算子，而多模态侧重图片、文本、音视频的原生类型支持、GPU加速推理、大模型推理等。

EMR Ray & Daft性能优化包含三方面。一是拓展Fusion边界，把关系算子的优化复制到Ray Data和Daft，包括向量化算子，对接Celeborn，Query Optimizer，Pipeline多线程，磁盘溢写等。二是提升GPU利用率，方法包括自动拆解和异步化CPU和GPU算子/UDF，自动扩容CPU避免GPU饥饿，自适应显存超卖等。三是提升调用大模型服务的优化，包括异步化，Batch化，多层次QO优化等，同时降低Token消耗和作业e2e延迟。

EMR Ray提供Ray Cluster和Ray Job两种作业提交形态，前者复用常驻集群最大化降低冷启动时间，适用于环境复用、低延迟场景；后者按作业申请资源，适用于环境不统一、稳定性要求高的批处理场景。

EMR Ray稳定高效支撑了多种工作负载，包括视频切割，抽帧，图片打标，图片向量化，文本生成等场景。

Spark & Ray统一数据底座

Spark和Ray构建在统一的数据管理底座之上，既支持开源开放的Hive Metastore(HMS) + OSS架构，也支持全托管DLF方案，既覆盖Paimon，Iceberg，Delta，Hudi等分钟级新鲜度的湖格式，也支持Fluss这种新兴的秒级流存储。

数据底座为Spark和Ray提供统一的全模态数据存储和管理服务，避免数据冗余。以Paimon为例，结构化和多模态数据分别以结构化和Blob类型存储，向量数据以Vector类型存储，同时提供标量和向量索引文件。Spark和Ray在同一份Paimon数据上读取、加工、生成、检索数据，互为上下游，真正做到一份数据，多引擎平权。

Spark & Ray统一算力和基础设施

Spark和Ray共用一份算力池，工作空间的CPU和GPU算力按需、细粒度在两个引擎之间统一高效供应，Spark释放的资源能立即被Ray消费，反之亦然，从而消除资源碎片，最大化利用率。

除了算力池，Spark和Ray还共享其他基础设施，包括统一的认证鉴权体系，统一的Cache服务，挂载CPFS、NAS、OSS的能力，监控报警等。整体架构如下所示：

Spark & Ray融合计算

Spark和Ray的融合计算有两种方式。一是作为数据管线的不同节点，互为依赖处理不同模态的数据，并以持久化的表或Raw Data作为传输介质。二是依赖RayDP，让Spark运行在Ray上，通过Ray Object Store实现Spark Dataframe和Ray Dataset之间的零拷贝互转，从而在一个作业里同时运行Spark和Ray。

案例

案例介绍

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

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

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

代码示例

多模态数据处理并非单一算子问题。一个视频数据集往往需要经历解析、切分、质量过滤、内容去重、语义标注、向量化和格式转换。全托管 Ray 为 Daft、Data-Juicer 等面向 AI 数据而生的引擎与工具提供统一运行底座，使企业能够在不自建 Ray 基础设施的前提下使用 Python 多模态生态。

Daft 提供 Python DataFrame 与 SQL 接口，可将视频解码、图像处理、模型推理等 CPU/GPU 算子组织为流水线，并由 Ray 负责分布式资源调度。其面向多模态数据的流式执行与背压机制，有助于让 I/O、CPU 预处理和 GPU 推理并行衔接，并支持 Paimon 等开放数据格式。

import daft
from daft import col
from daft.functions import encode_image
from daft.emr.functions import emr_udf
from daft.emr.functions import ai_query
from daft import lit

# 流式读取 OSS 视频，仅提取关键帧，每2秒采样一次
df = daft.read_video_frames(
    path="oss://<your-bucket>/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,
)

# 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")),
)

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

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()

结语

阿里云 EMR Serverless Spark 业内首发全托管 Spark + Ray 集群，通过极致内核优化和统一数据、算力底座，彻底打通了大数据工程与 AI 模型训练的割裂。结合 Daft、Data-Juicer 等多模态引擎，以及 CPFS、OSS 等高性能存储生态，阿里云正在为全球的 AI 开发者提供一套最具竞争力的数据新基建。