当前位置：首页 > news >正文

EMR Serverless Spark 数据湖上新能力：一条 SQL 实现标量向量混合检索

news 2026/6/6 1:56:14

摘要

功能定义：EMR Serverless Spark 原生集成标量-向量混合检索能力，基于 DLF Global Index 实现单一 SQL 接口下的多维数据查询。 **技术实现：**通过 vector_search UDF 结合标准 WHERE 子句，协同调用 Lumina 向量索引与 B-tree 标量索引，实现算子下推与联合执行。

核心优势：SQL 原生语法零门槛接入，DLF 索引全自动托管免运维，Serverless 存算分离按需计费无闲置。内置Celeborn 集中式 Shuffle 管理与 Fusion引擎，相比开源Spark 3倍性能提升。

架构价值：实现湖内原位检索（In-Lake Search），保障数据强一致性与低延迟。

典型场景：适用于智能驾驶长尾场景挖掘、多模态电商搜索、RAG 知识库增强及大规模离线向量批处理任务。

在海量数据中找到"看起来像"且"符合条件"的那条记录，过去需要两套系统协作——向量数据库管语义相似，关系型数据库管结构化过滤，数据在两者之间反复搬运，链路长、成本高、一致性难保障。

阿里云 EMR Serverless Spark 现已支持标量向量混合检索——基于 DLF Global Index 提供的向量索引能力，Spark SQL 原生支持标量过滤 + 向量近邻检索的混合查询，一条 SQL 即可完成"语义相似 + 条件约束"的双重检索。

一、什么是标量向量混合检索？为什么传统架构搞不定？

传统的数据检索面临一个两难：

纯标量查询（WHERE weather=‘暴雨’ AND speed>80）：精确，但找不到"语义相似"的场景
纯向量查询（Top-K 近邻）：能找"看起来像"的，但无法约束业务条件

以自动驾驶场景为例，工程师常常需要这样的查询：

“从历史数据中，找到所有天气为暴雨、道路类型为城市道路的、与当前场景最相似的 Top-10 个历史案例”

这种需求在传统架构下需要两步走：先从向量数据库取 Top-K，再在业务数据库中过滤——两次查询、数据搬运、结果可能不满足 K 条。

混合检索的意义：在向量近邻搜索的同时，增加标量过滤条件，一步到位，精确返回既相似又合规的结果。

二、阿里云 EMR Serverless Spark 如何实现标量向量混合检索？

阿里云 EMR Serverless Spark 基于数据湖内 DLF Paimon 表的 Global Index 能力，将向量索引和 B-tree 索引统一纳入 Spark SQL 的查询执行框架，实现了标量过滤与向量近邻检索的联合执行。

核心架构

两种索引可以同时建立在同一张 Paimon 表上，Spark 在查询时自动协同两路索引，无需用户关心底层调度。

向量索引的构建

在 Paimon 表上开启向量索引，只需在建表时指定表属性，Spark 作业写入数据后索引自动生效：

CREATETABLEai_dataset.scene_vectors(idBIGINT,path STRING,weather STRING,road_type STRING,speed_range STRING,embedding ARRAY<FLOAT>)USINGpaimon TBLPROPERTIES(-- 启用 Global Index 基础能力'row-tracking.enabled'='true','data-evolution.enabled'='true',-- 开启向量索引'morax.lumina-index.enabled'='true','global-index.lumina.index-column'='embedding',-- 指定向量维度（须与实际 embedding 维度一致）'lumina.index.dimension'='1152');

索引构建时机：

自动构建：在 TBLPROPERTIES 中声明morax.*相关参数，DLF 会根据表中的向量列自动调度构建索引。
手动触发：如果写入数据后希望立即创建索引，或建表时未声明morax.*相关参数，可以手动调用触发构建索引。

vector_search：Spark SQL 中的向量检索函数

Spark SQL 新增vector_search表函数，支持直接在 SQL 中进行向量近邻检索：

-- 基础向量检索：找到与查询向量最相似的 Top-K 条记录SELECT*FROMvector_search('ai_dataset.scene_vectors',-- 目标表'embedding',-- 向量列array(0.12F,0.34F,...),-- 查询向量10-- Top-K);

标量向量混合检索：一步到位

将vector_search的结果与标量 WHERE 条件结合，即可实现混合检索：

-- 混合检索：天气=暴雨 AND 城市道路 的 Top-10 相似场景SELECTid,path,weather,road_type,speed_rangeFROMvector_search('ai_dataset.scene_vectors','embedding',array(0.12F,0.34F,...),-- 当前场景的 embedding10)WHEREweather='heavy_rain'ANDroad_type='urban';

执行逻辑：Spark 通过向量索引检索近邻候选集，同时通过 B-tree 索引对标量条件进行过滤，两路协同、一步完成，无需跨系统数据搬运。

三、实践：智能驾驶场景召回，为模型训练准备数据集

感知模型在恶劣天气下表现不佳，需要大量特定条件的场景数据来重新训练和微调模型。但路采数据中这类场景占比极低，人工逐帧筛选效率极差。

以下演示如何用阿里云 EMR Serverless Spark SQL 完成"场景数据入湖 → 混合检索召回 → 导出训练集"的完整流程。

Step 1：路采数据入湖，AI 自动生成标签和向量

路测车每天产生大量视频帧，存储在 OSS 上。通过阿里云 EMR Serverless Spark AI Function，一条 SQL 完成图片读取、标签提取和向量化：

-- 创建场景表（含向量索引和 B-tree 索引）CREATETABLEIFNOTEXISTSad_dataset.driving_scenes(idBIGINT,path STRING,-- OSS 图片路径weather STRING,-- 天气：sunny/cloudy/rainy/snowy/foggy/otherlighting STRING,-- 光照：daytime/nighttime/dusk/tunnel/otherroad_type STRING,-- 道路类型：urban/expressway/rural/...objects ARRAY<STRING>,-- 检测到的目标类别risks ARRAY<STRING>,-- 检测到的风险类别scene_tag STRING,-- 场景标签（固定值）sensor_type STRING,-- 传感器类型（固定值）embedding ARRAY<FLOAT>-- 图片向量（1152 维）)USINGpaimon TBLPROPERTIES('row-tracking.enabled'='true','data-evolution.enabled'='true','morax.lumina-index.enabled'='true','global-index.lumina.index-column'='embedding','lumina.index.dimension'='1152','global-index.btree.index-columns'='weather,road_type,lighting,objects,risks,scene_tag');-- 批量入湖：读取 OSS 图片 → AI 生成标签 + 向量 → 写入 PaimonWITHrawAS(SELECTmonotonically_increasing_id()ASid,path,ai_query('You are an autonomous driving data analysis assistant. '||'Based on the input road scene image, output a JSON object with the following structure: '||'{"weather": "sunny/cloudy/rainy/snowy/foggy/other", '||'"lighting": "daytime/nighttime/dusk/tunnel/other", '||'"road_type": "urban/expressway/rural/intersection/ramp/parking_lot/other", '||'"objects": ["car", "pedestrian", "bicycle", "motorcycle", "bus", "truck", '||'"traffic_light", "traffic_sign", "cone", "construction_equipment"], '||'"risks": ["construction", "congestion", "occlusion", "accident_signs", '||'"wrong_way", "illegal_parking", "water_logging", "ice", "other"]}. '||'All field values must exactly match the enum options above (case-sensitive). '||'objects and risks must be arrays of strings; use empty array [] if none. '||'Do not output any extra text, explanation, Markdown, or code block. '||'Output only valid JSON.',data=>content)ASscene_json,ai_embedding_multimodal(content)ASembeddingFROMread_files('oss://ad-team-raw/camera_front/2025-*/',suffix=>'jpg,png'))INSERTINTOad_dataset.driving_scenesSELECTid,path,get_json_object(scene_json,'$.weather')ASweather,get_json_object(scene_json,'$.lighting')ASlighting,get_json_object(scene_json,'$.road_type')ASroad_type,from_json(get_json_object(scene_json,'$.objects'),'ARRAY<STRING>')ASobjects,from_json(get_json_object(scene_json,'$.risks'),'ARRAY<STRING>')ASrisks,'normal'ASscene_tag,'camera_front'ASsensor_type,embeddingFROMraw;

一条 INSERT 完成三件事：读取 OSS 图片 → AI 函数自动打标 + 生成向量 → 写入带双索引的 Paimon 表，后续查询即可利用向量索引和 B-tree 索引加速。

Step 2：混合检索召回目标场景

感知模型在"暴雨+城市道路"场景下误判率高，需要从历史库中召回相似场景用于模型重训练：

##以某次典型误判场景的 embedding 为查询向量query_vec=spark.sql(""" SELECT embedding FROM ad_dataset.driving_scenes WHERE path = 'oss://ad-team-raw/camera_front/2025-10-15/frame_CF_003812.jpg' """).collect()[0]["embedding"]vec_literal="array("+",".join(f"{v}f"forvinquery_vec)+")"#召回"暴雨+城市道路"下与典型误判场景最相似的历史案例result=spark.sql(f""" SELECT id, path, weather, road_type, lighting, objects, risks FROM vector_search( 'ad_dataset.driving_scenes', 'embedding',{vec_literal}, 500 ) WHERE weather = 'rainy' AND road_type = 'urban' AND lighting = 'nighttime' """)result.show(truncate=False)

混合检索的优势：向量检索先从海量数据中找到语义相似的场景，B-tree 索引再精准过滤天气、道路、时段等条件。两路协同，一步到位——传统方案需要先从向量库取候选、再在业务库中二次筛选，且无法保证结果数量。

Step 3：批量召回多种 Corner Case，构建训练集

模型训练不止需要一种场景。用 Spark SQL 的批处理能力，一次作业批量召回多种恶劣场景，直接写入训练数据集：

fromfunctoolsimportreducefrompyspark.sqlimportDataFrame# 批量召回多种恶劣场景，SCENES=[("oss://ad-team-raw/camera_front/2025-10-15/frame_CF_003812.jpg",{"weather":"rainy","road_type":"urban"},500,"rainy_urban",),("oss://ad-team-raw/camera_front/2025-11-02/frame_CF_001547.jpg",{"weather":"foggy","road_type":"expressway"},500,"foggy_expressway",),("oss://ad-team-raw/camera_front/2025-12-08/frame_CF_000923.jpg",{"weather":"snowy","road_type":"rural","lighting":"nighttime"},300,"snowy_rural_nighttime",),]defrecall_scene(anchor_path:str,filters:dict,top_k:int,source_label:str)->DataFrame:vec=spark.sql(f""" SELECT embedding FROM ad_dataset.driving_scenes WHERE path = '{anchor_path}' """).collect()[0]["embedding"]vec_literal="array("+",".join(f"{v}f"forvinquery_vec)+")"where_clause=" AND ".join(f"{col}= '{val}'"forcol,valinfilters.items())# 执行向量召回并附加场景标签returnspark.sql(f""" SELECT id, path, weather, lighting, road_type, objects, risks, scene_tag, sensor_type, '{source_label}' AS source_query FROM vector_search( 'ad_dataset.driving_scenes', 'embedding',{vec_literal},{top_k}) WHERE{where_clause}""")# 逐场景召回并合并结果frames=[recall_scene(*scene)forsceneinSCENES]result=reduce(DataFrame.unionByName,frames)# 将结果写入训练集spark.sql(""" CREATE TABLE IF NOT EXISTS ad_dataset.training_set_corner_cases ( id BIGINT, path STRING, weather STRING, lighting STRING, road_type STRING, objects ARRAY<STRING>, risks ARRAY<STRING>, scene_tag STRING, sensor_type STRING, source_query STRING ) USING paimon """)result.writeTo("ad_dataset.training_set_corner_cases").append()print(f"Total records written:{result.count()}")

关键价值：一条 SQL 同时完成多种场景的召回 + 合并，source_query字段标注了每条样本的来源召回条件，便于训练时按场景加权采样。

Step 4：训练集质量分析与去重

召回后需要分析训练集分布、去除重复样本——这些都可以在 Spark SQL 中直接完成：

-- 分析各场景的召回数量分布SELECTsource_query,COUNT(*)ASsample_countFROMad_dataset.training_set_corner_casesGROUPBYsource_queryORDERBYsample_countASC;-- 与维度表 JOIN，统计不同城市的覆盖情况SELECTt.source_query,r.city,COUNT(*)ASsample_countFROMad_dataset.training_set_corner_cases tJOINdim_road_info rONt.path=r.image_pathGROUPBYt.source_query,r.cityHAVINGCOUNT(*)>5ORDERBYt.source_query,sample_countDESC;