VECTOR_SEARCH_AGG-实时计算 Flink版(Flink)-阿里云帮助中心

本文介绍了如何通过 VECTOR_SEARCH_AGG 函数使用向量搜索的功能。向量搜索允许根据给定的高维数字向量以查找语义上最相似的项。相较于VECTOR_SEARCH，本函数以数组形式返回所有匹配项。

使用限制

实时计算Flink版VVR引擎11.3及以上版本支持流模式，11.4及以上版本支持批模式。
OUTPUT_COLUMNS 参数从实时计算Flink版VVR引擎11.9及以上版本支持。
仅支持 Milvus 作为向量表进行搜索。
仅支持处理非更新流（只包含INSERT类型消息的称为非更新流）。

语法

VECTOR_SEARCH_AGG(
  SEARCH_TABLE => TABLE <SEARCH_TABLE>,
  COLUMN_TO_SEARCH => DESCRIPTOR(<COLUMN_TO_SEARCH>),
  COLUMN_TO_QUERY => <COLUMN_TO_QUERY>,
  TOP_K => <TOP_K>
  [, CONFIG => <CONFIG>]
  [, OUTPUT_COLUMNS => DESCRIPTOR(<OUTPUT_COLUMN>[, ...])]
)

入参

参数	数据类型	说明
TABLE <SEARCH_TABLE>	TABLE	向量表的表名。
DESCRIPTOR(<COLUMN_TO_SEARCH>)	DESC	向量表中建立索引的向量列。输入数据将会跟指定的向量列比较，并计算相似度。
COLUMN_TO_QUERY	ARRAY<FLOAT>/ARRAY<DOUBLE>	输入中的向量特征列（例如用户上传的图像/文本的嵌入向量），并与建立索引的向量列进行相似度匹配。
TOP_K	INT	输出相似数据的最大条数。
CONFIG	MAP<STRING,STRING>	可配置的运行参数。
OUTPUT_COLUMNS	DESC	可选。指定数组元素中需要返回的向量搜索结果字段，可以选择向量表中的顶层列以及引擎生成的相似度字段 `score`。字段顺序与 `DESCRIPTOR` 中的声明顺序一致。不指定时返回向量表的所有字段和 `score`。

VECTOR_SEARCH_AGG 返回一张只有一条记录的数据表，表结构为 ROW<ARRAY>。

数组列名为 search_results。未指定 OUTPUT_COLUMNS 时，其中每个 ROW 都包含向量表中的所有列以及一个额外的 score 字段，与此前版本的行为一致。指定 OUTPUT_COLUMNS 时，数组元素中的每个 ROW 仅包含指定字段，并按照 DESCRIPTOR 中的顺序排列。

score 字段为 DOUBLE 类型，用于描述输入数据与搜索结果的相似度，也可以通过 OUTPUT_COLUMNS 选择或裁剪。如果向量表中已经存在名为 score 的列，引擎会为生成的相似度字段选择唯一名称，例如 score0；此时应在 OUTPUT_COLUMNS 中使用实际生成的字段名。

OUTPUT_COLUMNS 不能为空、不支持嵌套字段，并且不能包含重复字段。

运行参数

参数	数据类型	默认值	说明
async	Boolean	（none）	是否启用 async 模式。如果向量表使用的 connector 不支持指定的模式，引擎会报错告知用户。默认情况下，引擎会根据连接器支持的模式来选择执行模式。若连接器同时支持异步和同步模式，引擎优先使用异步模式提升整体的吞吐。
max-concurrent-operations	Integer	10	异步模式下，最大并行请求数。
output-mode	Enum	ORDERED	异步操作的输出模式。可能的取值： ORDERED ALLOW_UNORDERED 关于两个取值的含义，请参考Async I/O。
timeout	Duration	3min	从首次调用到异步操作最终完成的超时时间，可能包含多次重试，并且在发生故障转移（failover）时会被重置。

示例

测试数据

假设向量表 vector_table中有如下数据：

id	topic	vector_index
1	'BigData'	[1, 1, 0]
2	'Streaming'	[-5, -12, -13]
3	'Batch'	[5, 12, 13]

假设查询数据表 query_table中有如下数据：

id	user_keyword	embedding
1	'Spark'	[5, 12, 13]
2	'Flink'	[-5, -12, -13]

测试语句

通过以下 SQL 为 query_table中的每条数据搜索vector_table获取相似度前2的数据。

SELECT user_keyword, search_results
FROM
  query_table,
  LATERAL TABLE (VECTOR_SEARCH_AGG(
    SEARCH_TABLE => TABLE vector_table,
    COLUMN_TO_SEARCH => DESCRIPTOR(vector_index),
    COLUMN_TO_QUERY => query_table.embedding,
    TOP_K => 2,
    CONFIG => MAP['async', 'false'], -- 开启同步模式
    OUTPUT_COLUMNS => DESCRIPTOR(topic, score)
    ))

输出结果

user_keyword	search_results
'Spark'	[('Batch', 1.0), ('BigData', 0.6538461538461539)]
'Flink'	[('Streaming', 1.0), ('BigData', -0.6538461538461539)]