在 Flink SQL 里做向量检索 VECTOR_SEARCH - 教程

1. 背景：为什么要在 Flink 里做向量搜索？

大模型和向量数据库火起来之后，向量检索基本成了各种智能应用的标配：

• RAG（检索增强生成）：用文本向量找最相关的文档片段；
• 推荐 / 相似内容：根据用户向量找相似用户，或者根据物料向量找相似物料；
• 画像 / embedding 检索：根据用户行为 embedding 找“最近邻”人群做实时推荐。

典型架构一般是：

1. Flink / 各种 ETL 把数据加工成特征；
2. 把文本或特征喂给 embedding 模型生成向量；
3. 把向量写入向量数据库（如 Milvus、pgvector、ES / OpenSearch vector 字段等）；
4. 查询阶段，应用服务拿到查询向量，调用向量库 API 获取 Top-K，再走后续逻辑。

这套架构有两个痛点：

• 实时性不好控制：实时流进 Flink，结果还要绕一圈到应用层才能做向量检索；
• 逻辑分裂：特征加工在 Flink，向量检索在服务层，SQL/代码分散在不同系统中。

Flink 提供的 VECTOR_SEARCH 表值函数，就是把这件事整合回 SQL 世界：

让你直接在 Flink SQL 查询里写：
FROM input_table, LATERAL TABLE(VECTOR_SEARCH(...))
一边在流上加工数据，一边实时查向量库，拿到 Top-K 相似结果和 score。

这对实时推荐、实时 RAG Pipeline、相似告警合并等场景非常友好。

2. VECTOR_SEARCH 是什么？和 lookup join 有什么关系？

直观理解：

• 它长得很像一个 特殊的 lookup join；

• 只不过匹配条件不是 ON a.key = b.key，而是：

  用输入行的向量和外部表里的向量算相似度，然后返回 Top-K 行。

在 SQL 里，它是一个 表值函数（TVF），通常和 LATERAL TABLE 一起使用：

FROM input_table,
LATERAL TABLE(VECTOR_SEARCH(...))

语义上相当于：

• 对 input_table 中的每一行拿到一个向量 v_in；
• 在 vector_table 中按某一列 index_column 做相似度检索；
• 返回 Top-K 相似行以及相似度 score；
• 再把这些行与原始行拼成一张“展开的”结果表。

3. 语法拆解：每一块到底干什么？

完整写法：

SELECT *
FROM input_table,
LATERAL TABLE(
VECTOR_SEARCH(
TABLE vector_table,
input_table.vector_column,
DESCRIPTOR(index_column),
top_k,
[CONFIG => MAP['key', 'value']]
)
);

3.1 input_table：查询“驱动表”

• 是流式输入：比如实时行为流、实时日志流；
• 每一条记录中会包含一个向量列 vector_column，或者可以先在 SQL 里把 embedding 拼成数组。

3.2 vector_table：向量索引所在的外部表

• 通常对应一个向量数据库 / 向量索引系统（Milvus、pgvector、ES 向量字段等等），
• Connector 侧需要实现 VectorSearchTableSource；
• SQL 里你只需要把它当成普通的 external table 来用。

3.3 input_table.vector_column：要用来检索的向量

• 类型必须是 FLOAT ARRAY 或 DOUBLE ARRAY；

• 你可以：

  • 直接从上游数据里拿 embedding；
  • 或者在 Flink SQL 里通过 UDF + 特征拼接成向量列。

举例：

-- 假设 user_embedding 是 FLOAT ARRAY
input_table.user_embedding

3.4 DESCRIPTOR(index_column)：向量表里的索引列

• 指定 vector_table 中哪一列是用来做相似度搜索的向量列；
• 一般会是向量数据库中的 embedding 列。

DESCRIPTOR(embedding)  -- vector_table.embedding

3.5 top_k：返回多少个最近邻

• 经典向量检索参数；
• 对每条输入记录，会返回 Top-K 相似行；
• 注意：结果表会“变宽也变高”，一条输入行可能展开成 K 条结果。

3.6 CONFIG：异步搜索配置

和 ML_PREDICT 一样，异步配置也有这些：

• async：

  • 'true' → 使用异步向量搜索 Provider；
  • 'false' → 使用同步；

• max-concurrent-operations：最多挂起多少个异步请求；

• output-mode：

  • "ORDERED"：按输入顺序输出；
  • "ALLOW_UNORDERED"：允许乱序；

• timeout：从第一次调用到向量搜索完成的超时。

示例：

MAP[
'async', 'true',
'max-concurrent-operations', '2000',
'output-mode', 'ALLOW_UNORDERED',
'timeout', '3s'
]

4. 典型使用示例

4.1 基础用法：实时查询相似物料

假设：

• user_stream：实时用户行为流，里面已经有用户 embedding；
• item_vectors：商品向量表，存放物料 embedding。

-- user_stream: 实时用户行为 + embedding
CREATE TABLE user_stream (
user_id     STRING,
action      STRING,
emb         FLOAT ARRAY,
ts          TIMESTAMP_LTZ(3),
WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (...);
-- item_vectors: 商品向量外部表
CREATE TABLE item_vectors (
item_id     STRING,
title       STRING,
desc        STRING,
emb         FLOAT ARRAY
) WITH (...); -- 实现了 VectorSearchTableSource 的 Connector
-- 实时查找 Top-10 相似商品
SELECT
u.user_id,
u.action,
i.item_id,
i.title,
v.score      -- 相似度
FROM user_stream AS u,
LATERAL TABLE(VECTOR_SEARCH(
TABLE item_vectors,
u.emb,                      -- 输入向量
DESCRIPTOR(emb),            -- 索引列
10                          -- top_k
)) AS v                        -- v 中包含 item_vectors 的列和 score
JOIN item_vectors AS i ON v.item_id = i.item_id;   -- 或直接从 v 中取

实际上 VECTOR_SEARCH(...) 的输出已经包含 item_vectors 的列，你可以不再单独 join 一次，这里只是形式上写清楚。

4.2 带异步配置的实时向量检索

SELECT
u.user_id,
i.item_id,
i.title,
v.score
FROM user_stream AS u,
LATERAL TABLE(VECTOR_SEARCH(
TABLE item_vectors,
u.emb,
DESCRIPTOR(emb),
10,
MAP[
'async', 'true',
'max-concurrent-operations', '2000',
'output-mode', 'ALLOW_UNORDERED',
'timeout', '2s'
]
)) AS v
JOIN item_vectors AS i ON v.item_id = i.item_id;

4.3 用常量向量直接搜索（不需要 LATERAL）

当查询向量是常量 / 字面量时，不需要 LATERAL：

SELECT *
FROM TABLE(VECTOR_SEARCH(
TABLE item_vectors,
ARRAY[0.12, 0.35, -0.11, ...],  -- 常量向量
DESCRIPTOR(emb),
5
));

这在做 一次性分析 / 调试 embedding 时很方便。

5. 输出结构与相似度 score

VECTOR_SEARCH 的输出表包含：

1. 输入表 input_table 的所有列；
2. 向量表 vector_table 的所有列；
3. 一个额外的 score 列。

你可以把它当成一种特殊 join 的结果，只不过 join 条件变成了“按相似度 Top-K”。

score 的具体含义（内积、余弦相似度、L2 距离的负数等）由底层向量引擎决定，通常：

• 越大代表越相似（比如内积 / 余弦相似度）；
• 或者越小越相似（L2 距离），此时 connector 可以做适配，转成“相似度分数”。

无论如何，在 SQL 里你可以对 score 再做二次处理：

• 过滤低于阈值的结果；
• 对同一 user / 查询向量做二次排序等。

6. 限制与实现要求

6.1 只支持 append-only 表

文档里和 ML_PREDICT 一样强调：

VECTOR_SEARCH 只支持消费追加表（append-only tables）。

原因类似：向量检索通常依赖外部索引结构，变更行（update / delete）会让语义变得很复杂，并且向量库本身的更新语义一般是异步 eventually consistent。

6.2 Connector 必须实现 VectorSearchTableSource

向量表 vector_table 背后的 Source 必须实现：

• org.apache.flink.table.connector.source.VectorSearchTableSource

这个接口定义了 Flink 如何向外部向量服务发起检索请求、如何接收 Top-K 结果。
你在 SQL 这一侧不用关心实现细节，只需要知道：

• 不是什么表都能拿来 VECTOR_SEARCH；
• 只有专门实现了这个接口的表才行。

7. 性能调优和实践建议

要让 VECTOR_SEARCH 在生产环境里跑得又稳又快，可以关注这几个点：

1. 优先使用异步模式（async = true）

   • 向量检索基本都是网络调用 + 索引扫描，单次延迟普遍较高；
   • 异步模式可以“挂起”很多并发请求，把网络延迟藏在算子内部。

2. 合理设置 max-concurrent-operations

   • 太小：并发不足，CPU/网络都吃不满；
   • 太大：向量库会被打爆，或者 Flink 任务本身内存 / 连接资源吃紧；
   • 一般可以按：(期望QPS * 远端平均延迟) 做一个量级估算，再慢慢调优。

3. 视业务情况选择 ORDERED / ALLOW_UNORDERED

   • 多数“Top-K 相似结果”的业务对“结果顺序 = 输入顺序”没有强依赖，可以用 ALLOW_UNORDERED 换一点性能；
   • 如果你需要保证“对每条输入记录，返回结果必须严格按输入顺序输出”，就用默认 ORDERED。

4. 在 Flink 内完成尽可能多的特征/向量预处理

   • 比如先在 SQL 里把多字段拼成一个向量，再调用 VECTOR_SEARCH；
   • 这样向量库只负责检索，而不是做额外的 feature 工程。

8. 小结

VECTOR_SEARCH 把“向量检索”这件事变成了 Flink SQL 里的一个普通 TVF：

• 语法上和你已经熟悉的 LATERAL TABLE(...)、ML_PREDICT、各种窗口 TVF 非常统一；
• 可以直接在 流式 SQL 任务 中，对实时事件做向量检索；
• 再配上 ML_PREDICT，可以把“特征 + 推理 + 向量检索”都收敛到一份 SQL 作业里。

一句话总结：

以前要写一堆服务代码调用向量库，现在可以在 Flink SQL 里一句
..., LATERAL TABLE(VECTOR_SEARCH(...)) 搞定。