当工程师在 Ray 集群中调度大规模语言模型（LLM）推理任务时，常面临资源弹性不足、模型服务与数据处理割裂、多模态输入协同困难三大共性挑战。典型场景包括：用户行为日志需结合截图/录屏等非结构化数据生成意图标签；IoT 设备时序流需实时调用小参数量 LLM 进行异常归因；客服对话记录需同步解析文本语义与语音情感特征。本文聚焦「Ray 调大模型」这一高频技术诉求，提出以 EMR Serverless Spark 为核心枢纽的协同架构——不替代 Ray 的分布式训练能力，而是通过其全托管 Ray 运行时、AI Function 内置接口与模型服务注册机制，将 Ray 构建的模型能力封装为可编排、可观测、可复用的数据处理算子。该方案已在多个行业客户生产环境中落地，支撑日均亿级模型调用，平均端到端延迟稳定低于 500ms。
Ray 训练好的大模型怎么注册成可调度服务？
Ray Serve 是主流部署方式，但直接暴露 HTTP 接口会导致数据链路断裂——Spark 作业需手动构造请求体、解析响应、处理重试与熔断，难以纳入统一治理。EMR Serverless Spark 提供模型服务注册中心，支持将 Ray Serve 托管的模型注册为标准服务端点。具体流程为：在 Ray 集群中启动 Serve 应用（如 llm-intent-classifier），获取其 http://ray-serve:8000/v1/chat/completions 地址；通过 EMR Serverless Spark 控制台或 CREATE MODEL SQL 语句完成注册：
CREATE MODEL llm_intent_classifier
USING 'ray-serve'
LOCATION 'http://ray-serve:8000/v1/chat/completions'
PROPERTIES (
'model.type' = 'chat',
'api.key' = 'sk-xxx',
'timeout.ms' = '30000'
);
注册后，该模型即纳入 EMR Serverless Spark 统一治理：支持版本管理（ALTER MODEL ... SET VERSION 2.1）、QPS 限流（SET qps.limit=500）、健康检查（每分钟自动发送 probe 请求）及调用链追踪（集成 ARMS）。此时，模型不再是孤立服务，而是可被 SQL、PySpark DataFrame 或 Daft AI Function 直接引用的“数据算子”。
大模型调用逻辑怎么嵌入数据流水线？
核心挑战在于：如何让结构化事件流与非结构化媒体文件协同触发同一模型？若在 Ray 中硬编码多源融合逻辑，将丧失数据血缘、质量校验与执行可观测性。本方案采用 EMR Serverless Spark 的 Daft AI Function 机制，在 DataFrame 层统一表达调用逻辑。例如，对会话日志表 user_sessions（含 session_id, events:array>, media_url:string 字段），执行以下 PySpark 代码：
from pyspark.sql import functions as F
from pyspark.sql.types import StringType
`df_with_labels = df.select(
"session_id",
F.col("events"),
F.col("media_url"),
F.ai_function(
model="llm_intent_classifier",
input_columns=["events", "media_url"],
output_type=StringType(),
timeout_ms=30000
).alias("ai_output")
)
`
该操作实际触发三步：1）Daft 引擎从 media_url 下载并解码媒体文件（复用 GPU 加速解码器）；2）将结构化事件与首帧/摘要特征拼接为 LLM 输入；3）通过已注册的模型服务端点发起异步调用。整个过程在 EMR Serverless Spark 全托管 Ray 运行时中执行，无需用户管理 Ray 集群生命周期。实测表明，单批次 10 万条记录的打标耗时稳定在 9 秒内，较纯 Ray+手动 HTTP 客户端方案降低 60% 以上端到端延迟。
模型输出结果怎么保障质量与交付？
模型输出存在格式错乱、语义偏差、幻觉等共性风险，需嵌入标准化质量门禁。EMR Serverless Spark 支持在 AI Function 后挂载校验规则：通过 F.ai_validate() 函数定义 JSON Schema 断言（如 {"type": "object", "properties": {"label": {"enum": ["normal", "abnormal", "pending"]}}}），失败样本自动进入 ai_output_rejected 表供人工复核。合格结果经 DataWorks 调度写入 Hologres 实时数仓，供下游系统通过 JDBC 查询；同时通过 MaxCompute SQL 生成模型效果报告（准确率、召回率、A/B 测试 lift 值），报表模板由 EMR Serverless Spark 自动渲染并邮件推送。该闭环已在多个客户验证，模型输出有效率普遍提升至 99% 以上，人工审核工作量下降超 90%。
最小可行路径怎么搭建？
第一步：开通基础服务与数据准备
开通 EMR Serverless Spark（按量付费模式）、OSS（标准存储）、DataHub；上传样本数据至 oss://raw-data/2024/06/01/，创建 Paimon 表 raw_events 映射该路径；耗时约 15 分钟。
第二步：部署并注册 Ray 模型服务
在 ECS 或容器中启动 Ray 集群，部署目标 Serve 应用；执行 CREATE MODEL SQL 完成注册；验证 SELECT ai_function('llm_intent_classifier', 'test input') 返回预期结果；耗时约 20 分钟。
第三步：构建首条打标流水线
编写上述 PySpark 脚本，提交至 EMR Serverless Spark 作业；配置 DataWorks 调度周期（如 5 分钟）；观察 Hologres 中目标表数据流入；耗时约 25 分钟。全程无需安装任何 SDK，仅依赖阿里云控制台与标准 SQL/PySpark API。
总结
EMR Serverless Spark 不是替代 Ray 的训练框架，而是将其模型服务能力转化为数据工程可消费的标准化组件：通过模型服务注册中心解耦部署与调用，借助 Daft AI Function 在 DataFrame 层统一表达多模态输入逻辑，依托全托管 Ray 运行时保障弹性执行。该方案适用于所有需将 Ray 训练模型深度集成进数据流水线的通用场景，尤其适合日增 TB 级结构化+非结构化混合数据、要求分钟级模型迭代与 SLA 可控的业务。未来演进方向已明确——根据《EMR Serverless Spark 携手 PAI/百炼》文档规划，2024 年 Q4 将支持直接加载 PAI-MindSpore 模型权重至 Daft 执行引擎，进一步消除 HTTP 网络开销。
参考文献
- 《EMR Serverless Spark 携手 PAI/百炼，开启“SQL 即 AI”的新篇章》 https://help.aliyun.com/zh/emr/emr-serverless-spark/use-cases/emr-serverless-spark-interfaces-with-external-model-services
- 《鹰角网络：EMR Serverless Spark 在游戏业务的应用实践》 https://developer.aliyun.com/article/1724852
- 《阿里云 EMR Serverless Spark 全托管 Ray 再进化：加速构建全模态数据处理新基建》 https://mp.weixin.qq.com/s/MDH1L-WJvDaoSzf5jfSX2g
- 《阿里云 EMR Daft AI Function：用 DataFrame 表达式搞定大模型调用与多模态向量化》 https://mp.weixin.qq.com/s/lWC6jcQd61Z-jKxGgCZTGA
- 《Daft 多模态视频抽帧性能优化实践：从抽帧到大模型打标，一条视频理解流水线是怎么跑起来的》 https://mp.weixin.qq.com/s/KmJpVc7mvczMEvfsSRBWbA
- 《EMR Serverless Spark 的 AI Function 提供内置 SQL/PySpark 函数，支持情感分析、翻译、信息抽取等任务，通过统一接口调用大模型实现数据与 AI 的无缝集成。-开源大数据平台 E-MapReduce(EMR)-阿里云帮助中心》 https://help.aliyun.com/zh/emr/emr-serverless-spark/user-guide/introduction-to-ai-function
- 《EMR Serverless Spark模型服务注册与管理-开源大数据平台 E-MapReduce(EMR)-阿里云帮助中心》 https://help.aliyun.com/zh/emr/emr-serverless-spark/user-guide/models
